--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Commit c873b9fa33ac187b14dd7fa857ee2f596d7124a8
Changes
14 files changed, 9 insertions(+), 6595 deletions(-)
Diff
diff --git a/sbapp/mqtt/__init__.py b/sbapp/mqtt/__init__.py
deleted file mode 100644
index 9372c8f8..00000000
--- a/sbapp/mqtt/__init__.py
+++ /dev/null
@@ -1,5 +0,0 @@
-__version__ = "2.1.1.dev0"
-
-
-class MQTTException(Exception):
- pass
diff --git a/sbapp/mqtt/client.py b/sbapp/mqtt/client.py
deleted file mode 100644
index 4ccc8696..00000000
--- a/sbapp/mqtt/client.py
+++ /dev/null
@@ -1,5004 +0,0 @@
-# Copyright (c) 2012-2019 Roger Light and others
-#
-# All rights reserved. This program and the accompanying materials
-# are made available under the terms of the Eclipse Public License v2.0
-# and Eclipse Distribution License v1.0 which accompany this distribution.
-#
-# The Eclipse Public License is available at
-# http://www.eclipse.org/legal/epl-v20.html
-# and the Eclipse Distribution License is available at
-# http://www.eclipse.org/org/documents/edl-v10.php.
-#
-# Contributors:
-# Roger Light - initial API and implementation
-# Ian Craggs - MQTT V5 support
-"""
-This is an MQTT client module. MQTT is a lightweight pub/sub messaging
-protocol that is easy to implement and suitable for low powered devices.
-"""
-from __future__ import annotations
-
-import base64
-import collections
-import errno
-import hashlib
-import logging
-import os
-import platform
-import select
-import socket
-import string
-import struct
-import threading
-import time
-import urllib.parse
-import urllib.request
-import uuid
-import warnings
-from typing import TYPE_CHECKING, Any, Callable, Dict, Iterator, List, NamedTuple, Sequence, Tuple, Union, cast
-
-from paho.mqtt.packettypes import PacketTypes
-
-from .enums import CallbackAPIVersion, ConnackCode, LogLevel, MessageState, MessageType, MQTTErrorCode, MQTTProtocolVersion, PahoClientMode, _ConnectionState
-from .matcher import MQTTMatcher
-from .properties import Properties
-from .reasoncodes import ReasonCode, ReasonCodes
-from .subscribeoptions import SubscribeOptions
-
-try:
- from typing import Literal
-except ImportError:
- from typing_extensions import Literal # type: ignore
-
-if TYPE_CHECKING:
- try:
- from typing import TypedDict # type: ignore
- except ImportError:
- from typing_extensions import TypedDict
-
- try:
- from typing import Protocol # type: ignore
- except ImportError:
- from typing_extensions import Protocol # type: ignore
-
- class _InPacket(TypedDict):
- command: int
- have_remaining: int
- remaining_count: list[int]
- remaining_mult: int
- remaining_length: int
- packet: bytearray
- to_process: int
- pos: int
-
-
- class _OutPacket(TypedDict):
- command: int
- mid: int
- qos: int
- pos: int
- to_process: int
- packet: bytes
- info: MQTTMessageInfo | None
-
- class SocketLike(Protocol):
- def recv(self, buffer_size: int) -> bytes:
- ...
- def send(self, buffer: bytes) -> int:
- ...
- def close(self) -> None:
- ...
- def fileno(self) -> int:
- ...
- def setblocking(self, flag: bool) -> None:
- ...
-
-
-try:
- import ssl
-except ImportError:
- ssl = None # type: ignore[assignment]
-
-
-try:
- import socks # type: ignore[import-untyped]
-except ImportError:
- socks = None # type: ignore[assignment]
-
-
-try:
- # Use monotonic clock if available
- time_func = time.monotonic
-except AttributeError:
- time_func = time.time
-
-try:
- import dns.resolver
-
- HAVE_DNS = True
-except ImportError:
- HAVE_DNS = False
-
-
-if platform.system() == 'Windows':
- EAGAIN = errno.WSAEWOULDBLOCK # type: ignore[attr-defined]
-else:
- EAGAIN = errno.EAGAIN
-
-# Avoid linter complain. We kept importing it as ReasonCodes (plural) for compatibility
-_ = ReasonCodes
-
-# Keep copy of enums values for compatibility.
-CONNECT = MessageType.CONNECT
-CONNACK = MessageType.CONNACK
-PUBLISH = MessageType.PUBLISH
-PUBACK = MessageType.PUBACK
-PUBREC = MessageType.PUBREC
-PUBREL = MessageType.PUBREL
-PUBCOMP = MessageType.PUBCOMP
-SUBSCRIBE = MessageType.SUBSCRIBE
-SUBACK = MessageType.SUBACK
-UNSUBSCRIBE = MessageType.UNSUBSCRIBE
-UNSUBACK = MessageType.UNSUBACK
-PINGREQ = MessageType.PINGREQ
-PINGRESP = MessageType.PINGRESP
-DISCONNECT = MessageType.DISCONNECT
-AUTH = MessageType.AUTH
-
-# Log levels
-MQTT_LOG_INFO = LogLevel.MQTT_LOG_INFO
-MQTT_LOG_NOTICE = LogLevel.MQTT_LOG_NOTICE
-MQTT_LOG_WARNING = LogLevel.MQTT_LOG_WARNING
-MQTT_LOG_ERR = LogLevel.MQTT_LOG_ERR
-MQTT_LOG_DEBUG = LogLevel.MQTT_LOG_DEBUG
-LOGGING_LEVEL = {
- LogLevel.MQTT_LOG_DEBUG: logging.DEBUG,
- LogLevel.MQTT_LOG_INFO: logging.INFO,
- LogLevel.MQTT_LOG_NOTICE: logging.INFO, # This has no direct equivalent level
- LogLevel.MQTT_LOG_WARNING: logging.WARNING,
- LogLevel.MQTT_LOG_ERR: logging.ERROR,
-}
-
-# CONNACK codes
-CONNACK_ACCEPTED = ConnackCode.CONNACK_ACCEPTED
-CONNACK_REFUSED_PROTOCOL_VERSION = ConnackCode.CONNACK_REFUSED_PROTOCOL_VERSION
-CONNACK_REFUSED_IDENTIFIER_REJECTED = ConnackCode.CONNACK_REFUSED_IDENTIFIER_REJECTED
-CONNACK_REFUSED_SERVER_UNAVAILABLE = ConnackCode.CONNACK_REFUSED_SERVER_UNAVAILABLE
-CONNACK_REFUSED_BAD_USERNAME_PASSWORD = ConnackCode.CONNACK_REFUSED_BAD_USERNAME_PASSWORD
-CONNACK_REFUSED_NOT_AUTHORIZED = ConnackCode.CONNACK_REFUSED_NOT_AUTHORIZED
-
-# Message state
-mqtt_ms_invalid = MessageState.MQTT_MS_INVALID
-mqtt_ms_publish = MessageState.MQTT_MS_PUBLISH
-mqtt_ms_wait_for_puback = MessageState.MQTT_MS_WAIT_FOR_PUBACK
-mqtt_ms_wait_for_pubrec = MessageState.MQTT_MS_WAIT_FOR_PUBREC
-mqtt_ms_resend_pubrel = MessageState.MQTT_MS_RESEND_PUBREL
-mqtt_ms_wait_for_pubrel = MessageState.MQTT_MS_WAIT_FOR_PUBREL
-mqtt_ms_resend_pubcomp = MessageState.MQTT_MS_RESEND_PUBCOMP
-mqtt_ms_wait_for_pubcomp = MessageState.MQTT_MS_WAIT_FOR_PUBCOMP
-mqtt_ms_send_pubrec = MessageState.MQTT_MS_SEND_PUBREC
-mqtt_ms_queued = MessageState.MQTT_MS_QUEUED
-
-MQTT_ERR_AGAIN = MQTTErrorCode.MQTT_ERR_AGAIN
-MQTT_ERR_SUCCESS = MQTTErrorCode.MQTT_ERR_SUCCESS
-MQTT_ERR_NOMEM = MQTTErrorCode.MQTT_ERR_NOMEM
-MQTT_ERR_PROTOCOL = MQTTErrorCode.MQTT_ERR_PROTOCOL
-MQTT_ERR_INVAL = MQTTErrorCode.MQTT_ERR_INVAL
-MQTT_ERR_NO_CONN = MQTTErrorCode.MQTT_ERR_NO_CONN
-MQTT_ERR_CONN_REFUSED = MQTTErrorCode.MQTT_ERR_CONN_REFUSED
-MQTT_ERR_NOT_FOUND = MQTTErrorCode.MQTT_ERR_NOT_FOUND
-MQTT_ERR_CONN_LOST = MQTTErrorCode.MQTT_ERR_CONN_LOST
-MQTT_ERR_TLS = MQTTErrorCode.MQTT_ERR_TLS
-MQTT_ERR_PAYLOAD_SIZE = MQTTErrorCode.MQTT_ERR_PAYLOAD_SIZE
-MQTT_ERR_NOT_SUPPORTED = MQTTErrorCode.MQTT_ERR_NOT_SUPPORTED
-MQTT_ERR_AUTH = MQTTErrorCode.MQTT_ERR_AUTH
-MQTT_ERR_ACL_DENIED = MQTTErrorCode.MQTT_ERR_ACL_DENIED
-MQTT_ERR_UNKNOWN = MQTTErrorCode.MQTT_ERR_UNKNOWN
-MQTT_ERR_ERRNO = MQTTErrorCode.MQTT_ERR_ERRNO
-MQTT_ERR_QUEUE_SIZE = MQTTErrorCode.MQTT_ERR_QUEUE_SIZE
-MQTT_ERR_KEEPALIVE = MQTTErrorCode.MQTT_ERR_KEEPALIVE
-
-MQTTv31 = MQTTProtocolVersion.MQTTv31
-MQTTv311 = MQTTProtocolVersion.MQTTv311
-MQTTv5 = MQTTProtocolVersion.MQTTv5
-
-MQTT_CLIENT = PahoClientMode.MQTT_CLIENT
-MQTT_BRIDGE = PahoClientMode.MQTT_BRIDGE
-
-# For MQTT V5, use the clean start flag only on the first successful connect
-MQTT_CLEAN_START_FIRST_ONLY: CleanStartOption = 3
-
-sockpair_data = b"0"
-
-# Payload support all those type and will be converted to bytes:
-# * str are utf8 encoded
-# * int/float are converted to string and utf8 encoded (e.g. 1 is converted to b"1")
-# * None is converted to a zero-length payload (i.e. b"")
-PayloadType = Union[str, bytes, bytearray, int, float, None]
-
-HTTPHeader = Dict[str, str]
-WebSocketHeaders = Union[Callable[[HTTPHeader], HTTPHeader], HTTPHeader]
-
-CleanStartOption = Union[bool, Literal[3]]
-
-
-class ConnectFlags(NamedTuple):
- """Contains additional information passed to `on_connect` callback"""
-
- session_present: bool
- """
- this flag is useful for clients that are
- using clean session set to False only (MQTTv3) or clean_start = False (MQTTv5).
- In that case, if client that reconnects to a broker that it has previously
- connected to, this flag indicates whether the broker still has the
- session information for the client. If true, the session still exists.
- """
-
-
-class DisconnectFlags(NamedTuple):
- """Contains additional information passed to `on_disconnect` callback"""
-
- is_disconnect_packet_from_server: bool
- """
- tells whether this on_disconnect call is the result
- of receiving an DISCONNECT packet from the broker or if the on_disconnect is only
- generated by the client library.
- When true, the reason code is generated by the broker.
- """
-
-
-CallbackOnConnect_v1_mqtt3 = Callable[["Client", Any, Dict[str, Any], MQTTErrorCode], None]
-CallbackOnConnect_v1_mqtt5 = Callable[["Client", Any, Dict[str, Any], ReasonCode, Union[Properties, None]], None]
-CallbackOnConnect_v1 = Union[CallbackOnConnect_v1_mqtt5, CallbackOnConnect_v1_mqtt3]
-CallbackOnConnect_v2 = Callable[["Client", Any, ConnectFlags, ReasonCode, Union[Properties, None]], None]
-CallbackOnConnect = Union[CallbackOnConnect_v1, CallbackOnConnect_v2]
-CallbackOnConnectFail = Callable[["Client", Any], None]
-CallbackOnDisconnect_v1_mqtt3 = Callable[["Client", Any, MQTTErrorCode], None]
-CallbackOnDisconnect_v1_mqtt5 = Callable[["Client", Any, Union[ReasonCode, int, None], Union[Properties, None]], None]
-CallbackOnDisconnect_v1 = Union[CallbackOnDisconnect_v1_mqtt3, CallbackOnDisconnect_v1_mqtt5]
-CallbackOnDisconnect_v2 = Callable[["Client", Any, DisconnectFlags, ReasonCode, Union[Properties, None]], None]
-CallbackOnDisconnect = Union[CallbackOnDisconnect_v1, CallbackOnDisconnect_v2]
-CallbackOnLog = Callable[["Client", Any, int, str], None]
-CallbackOnMessage = Callable[["Client", Any, "MQTTMessage"], None]
-CallbackOnPreConnect = Callable[["Client", Any], None]
-CallbackOnPublish_v1 = Callable[["Client", Any, int], None]
-CallbackOnPublish_v2 = Callable[["Client", Any, int, ReasonCode, Properties], None]
-CallbackOnPublish = Union[CallbackOnPublish_v1, CallbackOnPublish_v2]
-CallbackOnSocket = Callable[["Client", Any, "SocketLike"], None]
-CallbackOnSubscribe_v1_mqtt3 = Callable[["Client", Any, int, Tuple[int, ...]], None]
-CallbackOnSubscribe_v1_mqtt5 = Callable[["Client", Any, int, List[ReasonCode], Properties], None]
-CallbackOnSubscribe_v1 = Union[CallbackOnSubscribe_v1_mqtt3, CallbackOnSubscribe_v1_mqtt5]
-CallbackOnSubscribe_v2 = Callable[["Client", Any, int, List[ReasonCode], Union[Properties, None]], None]
-CallbackOnSubscribe = Union[CallbackOnSubscribe_v1, CallbackOnSubscribe_v2]
-CallbackOnUnsubscribe_v1_mqtt3 = Callable[["Client", Any, int], None]
-CallbackOnUnsubscribe_v1_mqtt5 = Callable[["Client", Any, int, Properties, Union[ReasonCode, List[ReasonCode]]], None]
-CallbackOnUnsubscribe_v1 = Union[CallbackOnUnsubscribe_v1_mqtt3, CallbackOnUnsubscribe_v1_mqtt5]
-CallbackOnUnsubscribe_v2 = Callable[["Client", Any, int, List[ReasonCode], Union[Properties, None]], None]
-CallbackOnUnsubscribe = Union[CallbackOnUnsubscribe_v1, CallbackOnUnsubscribe_v2]
-
-# This is needed for typing because class Client redefined the name "socket"
-_socket = socket
-
-
-class WebsocketConnectionError(ConnectionError):
- """ WebsocketConnectionError is a subclass of ConnectionError.
-
- It's raised when unable to perform the Websocket handshake.
- """
- pass
-
-
-def error_string(mqtt_errno: MQTTErrorCode | int) -> str:
- """Return the error string associated with an mqtt error number."""
- if mqtt_errno == MQTT_ERR_SUCCESS:
- return "No error."
- elif mqtt_errno == MQTT_ERR_NOMEM:
- return "Out of memory."
- elif mqtt_errno == MQTT_ERR_PROTOCOL:
- return "A network protocol error occurred when communicating with the broker."
- elif mqtt_errno == MQTT_ERR_INVAL:
- return "Invalid function arguments provided."
- elif mqtt_errno == MQTT_ERR_NO_CONN:
- return "The client is not currently connected."
- elif mqtt_errno == MQTT_ERR_CONN_REFUSED:
- return "The connection was refused."
- elif mqtt_errno == MQTT_ERR_NOT_FOUND:
- return "Message not found (internal error)."
- elif mqtt_errno == MQTT_ERR_CONN_LOST:
- return "The connection was lost."
- elif mqtt_errno == MQTT_ERR_TLS:
- return "A TLS error occurred."
- elif mqtt_errno == MQTT_ERR_PAYLOAD_SIZE:
- return "Payload too large."
- elif mqtt_errno == MQTT_ERR_NOT_SUPPORTED:
- return "This feature is not supported."
- elif mqtt_errno == MQTT_ERR_AUTH:
- return "Authorisation failed."
- elif mqtt_errno == MQTT_ERR_ACL_DENIED:
- return "Access denied by ACL."
- elif mqtt_errno == MQTT_ERR_UNKNOWN:
- return "Unknown error."
- elif mqtt_errno == MQTT_ERR_ERRNO:
- return "Error defined by errno."
- elif mqtt_errno == MQTT_ERR_QUEUE_SIZE:
- return "Message queue full."
- elif mqtt_errno == MQTT_ERR_KEEPALIVE:
- return "Client or broker did not communicate in the keepalive interval."
- else:
- return "Unknown error."
-
-
-def connack_string(connack_code: int|ReasonCode) -> str:
- """Return the string associated with a CONNACK result or CONNACK reason code."""
- if isinstance(connack_code, ReasonCode):
- return str(connack_code)
-
- if connack_code == CONNACK_ACCEPTED:
- return "Connection Accepted."
- elif connack_code == CONNACK_REFUSED_PROTOCOL_VERSION:
- return "Connection Refused: unacceptable protocol version."
- elif connack_code == CONNACK_REFUSED_IDENTIFIER_REJECTED:
- return "Connection Refused: identifier rejected."
- elif connack_code == CONNACK_REFUSED_SERVER_UNAVAILABLE:
- return "Connection Refused: broker unavailable."
- elif connack_code == CONNACK_REFUSED_BAD_USERNAME_PASSWORD:
- return "Connection Refused: bad user name or password."
- elif connack_code == CONNACK_REFUSED_NOT_AUTHORIZED:
- return "Connection Refused: not authorised."
- else:
- return "Connection Refused: unknown reason."
-
-
-def convert_connack_rc_to_reason_code(connack_code: ConnackCode) -> ReasonCode:
- """Convert a MQTTv3 / MQTTv3.1.1 connack result to `ReasonCode`.
-
- This is used in `on_connect` callback to have a consistent API.
-
- Be careful that the numeric value isn't the same, for example:
-
- >>> ConnackCode.CONNACK_REFUSED_SERVER_UNAVAILABLE == 3
- >>> convert_connack_rc_to_reason_code(ConnackCode.CONNACK_REFUSED_SERVER_UNAVAILABLE) == 136
-
- It's recommended to compare by names
-
- >>> code_to_test = ReasonCode(PacketTypes.CONNACK, "Server unavailable")
- >>> convert_connack_rc_to_reason_code(ConnackCode.CONNACK_REFUSED_SERVER_UNAVAILABLE) == code_to_test
- """
- if connack_code == ConnackCode.CONNACK_ACCEPTED:
- return ReasonCode(PacketTypes.CONNACK, "Success")
- if connack_code == ConnackCode.CONNACK_REFUSED_PROTOCOL_VERSION:
- return ReasonCode(PacketTypes.CONNACK, "Unsupported protocol version")
- if connack_code == ConnackCode.CONNACK_REFUSED_IDENTIFIER_REJECTED:
- return ReasonCode(PacketTypes.CONNACK, "Client identifier not valid")
- if connack_code == ConnackCode.CONNACK_REFUSED_SERVER_UNAVAILABLE:
- return ReasonCode(PacketTypes.CONNACK, "Server unavailable")
- if connack_code == ConnackCode.CONNACK_REFUSED_BAD_USERNAME_PASSWORD:
- return ReasonCode(PacketTypes.CONNACK, "Bad user name or password")
- if connack_code == ConnackCode.CONNACK_REFUSED_NOT_AUTHORIZED:
- return ReasonCode(PacketTypes.CONNACK, "Not authorized")
-
- return ReasonCode(PacketTypes.CONNACK, "Unspecified error")
-
-
-def convert_disconnect_error_code_to_reason_code(rc: MQTTErrorCode) -> ReasonCode:
- """Convert an MQTTErrorCode to Reason code.
-
- This is used in `on_disconnect` callback to have a consistent API.
-
- Be careful that the numeric value isn't the same, for example:
-
- >>> MQTTErrorCode.MQTT_ERR_PROTOCOL == 2
- >>> convert_disconnect_error_code_to_reason_code(MQTTErrorCode.MQTT_ERR_PROTOCOL) == 130
-
- It's recommended to compare by names
-
- >>> code_to_test = ReasonCode(PacketTypes.DISCONNECT, "Protocol error")
- >>> convert_disconnect_error_code_to_reason_code(MQTTErrorCode.MQTT_ERR_PROTOCOL) == code_to_test
- """
- if rc == MQTTErrorCode.MQTT_ERR_SUCCESS:
- return ReasonCode(PacketTypes.DISCONNECT, "Success")
- if rc == MQTTErrorCode.MQTT_ERR_KEEPALIVE:
- return ReasonCode(PacketTypes.DISCONNECT, "Keep alive timeout")
- if rc == MQTTErrorCode.MQTT_ERR_CONN_LOST:
- return ReasonCode(PacketTypes.DISCONNECT, "Unspecified error")
- return ReasonCode(PacketTypes.DISCONNECT, "Unspecified error")
-
-
-def _base62(
- num: int,
- base: str = string.digits + string.ascii_letters,
- padding: int = 1,
-) -> str:
- """Convert a number to base-62 representation."""
- if num < 0:
- raise ValueError("Number must be positive or zero")
- digits = []
- while num:
- num, rest = divmod(num, 62)
- digits.append(base[rest])
- digits.extend(base[0] for _ in range(len(digits), padding))
- return ''.join(reversed(digits))
-
-
-def topic_matches_sub(sub: str, topic: str) -> bool:
- """Check whether a topic matches a subscription.
-
- For example:
-
- * Topic "foo/bar" would match the subscription "foo/#" or "+/bar"
- * Topic "non/matching" would not match the subscription "non/+/+"
- """
- matcher = MQTTMatcher()
- matcher[sub] = True
- try:
- next(matcher.iter_match(topic))
- return True
- except StopIteration:
- return False
-
-
-def _socketpair_compat() -> tuple[socket.socket, socket.socket]:
- """TCP/IP socketpair including Windows support"""
- listensock = socket.socket(
- socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_IP)
- listensock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
- listensock.bind(("127.0.0.1", 0))
- listensock.listen(1)
-
- iface, port = listensock.getsockname()
- sock1 = socket.socket(
- socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_IP)
- sock1.setblocking(False)
- try:
- sock1.connect(("127.0.0.1", port))
- except BlockingIOError:
- pass
- sock2, address = listensock.accept()
- sock2.setblocking(False)
- listensock.close()
- return (sock1, sock2)
-
-
-def _force_bytes(s: str | bytes) -> bytes:
- if isinstance(s, str):
- return s.encode("utf-8")
- return s
-
-
-def _encode_payload(payload: str | bytes | bytearray | int | float | None) -> bytes|bytearray:
- if isinstance(payload, str):
- return payload.encode("utf-8")
-
- if isinstance(payload, (int, float)):
- return str(payload).encode("ascii")
-
- if payload is None:
- return b""
-
- if not isinstance(payload, (bytes, bytearray)):
- raise TypeError(
- "payload must be a string, bytearray, int, float or None."
- )
-
- return payload
-
-
-class MQTTMessageInfo:
- """This is a class returned from `Client.publish()` and can be used to find
- out the mid of the message that was published, and to determine whether the
- message has been published, and/or wait until it is published.
- """
-
- __slots__ = 'mid', '_published', '_condition', 'rc', '_iterpos'
-
- def __init__(self, mid: int):
- self.mid = mid
- """ The message Id (int)"""
- self._published = False
- self._condition = threading.Condition()
- self.rc: MQTTErrorCode = MQTTErrorCode.MQTT_ERR_SUCCESS
- """ The `MQTTErrorCode` that give status for this message.
- This value could change until the message `is_published`"""
- self._iterpos = 0
-
- def __str__(self) -> str:
- return str((self.rc, self.mid))
-
- def __iter__(self) -> Iterator[MQTTErrorCode | int]:
- self._iterpos = 0
- return self
-
- def __next__(self) -> MQTTErrorCode | int:
- return self.next()
-
- def next(self) -> MQTTErrorCode | int:
- if self._iterpos == 0:
- self._iterpos = 1
- return self.rc
- elif self._iterpos == 1:
- self._iterpos = 2
- return self.mid
- else:
- raise StopIteration
-
- def __getitem__(self, index: int) -> MQTTErrorCode | int:
- if index == 0:
- return self.rc
- elif index == 1:
- return self.mid
- else:
- raise IndexError("index out of range")
-
- def _set_as_published(self) -> None:
- with self._condition:
- self._published = True
- self._condition.notify()
-
- def wait_for_publish(self, timeout: float | None = None) -> None:
- """Block until the message associated with this object is published, or
- until the timeout occurs. If timeout is None, this will never time out.
- Set timeout to a positive number of seconds, e.g. 1.2, to enable the
- timeout.
-
- :raises ValueError: if the message was not queued due to the outgoing
- queue being full.
-
- :raises RuntimeError: if the message was not published for another
- reason.
- """
- if self.rc == MQTT_ERR_QUEUE_SIZE:
- raise ValueError('Message is not queued due to ERR_QUEUE_SIZE')
- elif self.rc == MQTT_ERR_AGAIN:
- pass
- elif self.rc > 0:
- raise RuntimeError(f'Message publish failed: {error_string(self.rc)}')
-
- timeout_time = None if timeout is None else time_func() + timeout
- timeout_tenth = None if timeout is None else timeout / 10.
- def timed_out() -> bool:
- return False if timeout_time is None else time_func() > timeout_time
-
- with self._condition:
- while not self._published and not timed_out():
- self._condition.wait(timeout_tenth)
-
- if self.rc > 0:
- raise RuntimeError(f'Message publish failed: {error_string(self.rc)}')
-
- def is_published(self) -> bool:
- """Returns True if the message associated with this object has been
- published, else returns False.
-
- To wait for this to become true, look at `wait_for_publish`.
- """
- if self.rc == MQTTErrorCode.MQTT_ERR_QUEUE_SIZE:
- raise ValueError('Message is not queued due to ERR_QUEUE_SIZE')
- elif self.rc == MQTTErrorCode.MQTT_ERR_AGAIN:
- pass
- elif self.rc > 0:
- raise RuntimeError(f'Message publish failed: {error_string(self.rc)}')
-
- with self._condition:
- return self._published
-
-
-class MQTTMessage:
- """ This is a class that describes an incoming message. It is
- passed to the `on_message` callback as the message parameter.
- """
- __slots__ = 'timestamp', 'state', 'dup', 'mid', '_topic', 'payload', 'qos', 'retain', 'info', 'properties'
-
- def __init__(self, mid: int = 0, topic: bytes = b""):
- self.timestamp = 0.0
- self.state = mqtt_ms_invalid
- self.dup = False
- self.mid = mid
- """ The message id (int)."""
- self._topic = topic
- self.payload = b""
- """the message payload (bytes)"""
- self.qos = 0
- """ The message Quality of Service (0, 1 or 2)."""
- self.retain = False
- """ If true, the message is a retained message and not fresh."""
- self.info = MQTTMessageInfo(mid)
- self.properties: Properties | None = None
- """ In MQTT v5.0, the properties associated with the message. (`Properties`)"""
-
- def __eq__(self, other: object) -> bool:
- """Override the default Equals behavior"""
- if isinstance(other, self.__class__):
- return self.mid == other.mid
- return False
-
- def __ne__(self, other: object) -> bool:
- """Define a non-equality test"""
- return not self.__eq__(other)
-
- @property
- def topic(self) -> str:
- """topic that the message was published on.
-
- This property is read-only.
- """
- return self._topic.decode('utf-8')
-
- @topic.setter
- def topic(self, value: bytes) -> None:
- self._topic = value
-
-
-class Client:
- """MQTT version 3.1/3.1.1/5.0 client class.
-
- This is the main class for use communicating with an MQTT broker.
-
- General usage flow:
-
- * Use `connect()`, `connect_async()` or `connect_srv()` to connect to a broker
- * Use `loop_start()` to set a thread running to call `loop()` for you.
- * Or use `loop_forever()` to handle calling `loop()` for you in a blocking function.
- * Or call `loop()` frequently to maintain network traffic flow with the broker
- * Use `subscribe()` to subscribe to a topic and receive messages
- * Use `publish()` to send messages
- * Use `disconnect()` to disconnect from the broker
-
- Data returned from the broker is made available with the use of callback
- functions as described below.
-
- :param CallbackAPIVersion callback_api_version: define the API version for user-callback (on_connect, on_publish,...).
- This field is required and it's recommended to use the latest version (CallbackAPIVersion.API_VERSION2).
- See each callback for description of API for each version. The file docs/migrations.rst contains details on
- how to migrate between version.
-
- :param str client_id: the unique client id string used when connecting to the
- broker. If client_id is zero length or None, then the behaviour is
- defined by which protocol version is in use. If using MQTT v3.1.1, then
- a zero length client id will be sent to the broker and the broker will
- generate a random for the client. If using MQTT v3.1 then an id will be
- randomly generated. In both cases, clean_session must be True. If this
- is not the case a ValueError will be raised.
-
- :param bool clean_session: a boolean that determines the client type. If True,
- the broker will remove all information about this client when it
- disconnects. If False, the client is a persistent client and
- subscription information and queued messages will be retained when the
- client disconnects.
- Note that a client will never discard its own outgoing messages on
- disconnect. Calling connect() or reconnect() will cause the messages to
- be resent. Use reinitialise() to reset a client to its original state.
- The clean_session argument only applies to MQTT versions v3.1.1 and v3.1.
- It is not accepted if the MQTT version is v5.0 - use the clean_start
- argument on connect() instead.
-
- :param userdata: user defined data of any type that is passed as the "userdata"
- parameter to callbacks. It may be updated at a later point with the
- user_data_set() function.
-
- :param int protocol: allows explicit setting of the MQTT version to
- use for this client. Can be paho.mqtt.client.MQTTv311 (v3.1.1),
- paho.mqtt.client.MQTTv31 (v3.1) or paho.mqtt.client.MQTTv5 (v5.0),
- with the default being v3.1.1.
-
- :param transport: use "websockets" to use WebSockets as the transport
- mechanism. Set to "tcp" to use raw TCP, which is the default.
- Use "unix" to use Unix sockets as the transport mechanism; note that
- this option is only available on platforms that support Unix sockets,
- and the "host" argument is interpreted as the path to the Unix socket
- file in this case.
-
- :param bool manual_ack: normally, when a message is received, the library automatically
- acknowledges after on_message callback returns. manual_ack=True allows the application to
- acknowledge receipt after it has completed processing of a message
- using a the ack() method. This addresses vulnerability to message loss
- if applications fails while processing a message, or while it pending
- locally.
-
- Callbacks
- =========
-
- A number of callback functions are available to receive data back from the
- broker. To use a callback, define a function and then assign it to the
- client::
-
- def on_connect(client, userdata, flags, reason_code, properties):
- print(f"Connected with result code {reason_code}")
-
- client.on_connect = on_connect
-
- Callbacks can also be attached using decorators::
-
- mqttc = paho.mqtt.Client()
-
- @mqttc.connect_callback()
- def on_connect(client, userdata, flags, reason_code, properties):
- print(f"Connected with result code {reason_code}")
-
- All of the callbacks as described below have a "client" and an "userdata"
- argument. "client" is the `Client` instance that is calling the callback.
- userdata" is user data of any type and can be set when creating a new client
- instance or with `user_data_set()`.
-
- If you wish to suppress exceptions within a callback, you should set
- ``mqttc.suppress_exceptions = True``
-
- The callbacks are listed below, documentation for each of them can be found
- at the same function name:
-
- `on_connect`, `on_connect_fail`, `on_disconnect`, `on_message`, `on_publish`,
- `on_subscribe`, `on_unsubscribe`, `on_log`, `on_socket_open`, `on_socket_close`,
- `on_socket_register_write`, `on_socket_unregister_write`
- """
-
- def __init__(
- self,
- callback_api_version: CallbackAPIVersion = CallbackAPIVersion.VERSION1,
- client_id: str | None = "",
- clean_session: bool | None = None,
- userdata: Any = None,
- protocol: MQTTProtocolVersion = MQTTv311,
- transport: Literal["tcp", "websockets", "unix"] = "tcp",
- reconnect_on_failure: bool = True,
- manual_ack: bool = False,
- ) -> None:
- transport = transport.lower() # type: ignore
- if transport == "unix" and not hasattr(socket, "AF_UNIX"):
- raise ValueError('"unix" transport not supported')
- elif transport not in ("websockets", "tcp", "unix"):
- raise ValueError(
- f'transport must be "websockets", "tcp" or "unix", not {transport}')
-
- self._manual_ack = manual_ack
- self._transport = transport
- self._protocol = protocol
- self._userdata = userdata
- self._sock: SocketLike | None = None
- self._sockpairR: socket.socket | None = None
- self._sockpairW: socket.socket | None = None
- self._keepalive = 60
- self._connect_timeout = 5.0
- self._client_mode = MQTT_CLIENT
- self._callback_api_version = callback_api_version
-
- if self._callback_api_version == CallbackAPIVersion.VERSION1:
- warnings.warn(
- "Callback API version 1 is deprecated, update to latest version",
- category=DeprecationWarning,
- stacklevel=2,
- )
- if isinstance(self._callback_api_version, str):
- # Help user to migrate, it probably provided a client id
- # as first arguments
- raise ValueError(
- "Unsupported callback API version: version 2.0 added a callback_api_version, see docs/migrations.rst for details"
- )
- if self._callback_api_version not in CallbackAPIVersion:
- raise ValueError("Unsupported callback API version")
-
- self._clean_start: int = MQTT_CLEAN_START_FIRST_ONLY
-
- if protocol == MQTTv5:
- if clean_session is not None:
- raise ValueError('Clean session is not used for MQTT 5.0')
- else:
- if clean_session is None:
- clean_session = True
- if not clean_session and (client_id == "" or client_id is None):
- raise ValueError(
- 'A client id must be provided if clean session is False.')
- self._clean_session = clean_session
-
- # [MQTT-3.1.3-4] Client Id must be UTF-8 encoded string.
- if client_id == "" or client_id is None:
- if protocol == MQTTv31:
- self._client_id = _base62(uuid.uuid4().int, padding=22).encode("utf8")
- else:
- self._client_id = b""
- else:
- self._client_id = _force_bytes(client_id)
-
- self._username: bytes | None = None
- self._password: bytes | None = None
- self._in_packet: _InPacket = {
- "command": 0,
- "have_remaining": 0,
- "remaining_count": [],
- "remaining_mult": 1,
- "remaining_length": 0,
- "packet": bytearray(b""),
- "to_process": 0,
- "pos": 0,
- }
- self._out_packet: collections.deque[_OutPacket] = collections.deque()
- self._last_msg_in = time_func()
- self._last_msg_out = time_func()
- self._reconnect_min_delay = 1
- self._reconnect_max_delay = 120
- self._reconnect_delay: int | None = None
- self._reconnect_on_failure = reconnect_on_failure
- self._ping_t = 0.0
- self._last_mid = 0
- self._state = _ConnectionState.MQTT_CS_NEW
- self._out_messages: collections.OrderedDict[
- int, MQTTMessage
- ] = collections.OrderedDict()
- self._in_messages: collections.OrderedDict[
- int, MQTTMessage
- ] = collections.OrderedDict()
- self._max_inflight_messages = 20
- self._inflight_messages = 0
- self._max_queued_messages = 0
- self._connect_properties: Properties | None = None
- self._will_properties: Properties | None = None
- self._will = False
- self._will_topic = b""
- self._will_payload = b""
- self._will_qos = 0
- self._will_retain = False
- self._on_message_filtered = MQTTMatcher()
- self._host = ""
- self._port = 1883
- self._bind_address = ""
- self._bind_port = 0
- self._proxy: Any = {}
- self._in_callback_mutex = threading.Lock()
- self._callback_mutex = threading.RLock()
- self._msgtime_mutex = threading.Lock()
- self._out_message_mutex = threading.RLock()
- self._in_message_mutex = threading.Lock()
- self._reconnect_delay_mutex = threading.Lock()
- self._mid_generate_mutex = threading.Lock()
- self._thread: threading.Thread | None = None
- self._thread_terminate = False
- self._ssl = False
- self._ssl_context: ssl.SSLContext | None = None
- # Only used when SSL context does not have check_hostname attribute
- self._tls_insecure = False
- self._logger: logging.Logger | None = None
- self._registered_write = False
- # No default callbacks
- self._on_log: CallbackOnLog | None = None
- self._on_pre_connect: CallbackOnPreConnect | None = None
- self._on_connect: CallbackOnConnect | None = None
- self._on_connect_fail: CallbackOnConnectFail | None = None
- self._on_subscribe: CallbackOnSubscribe | None = None
- self._on_message: CallbackOnMessage | None = None
- self._on_publish: CallbackOnPublish | None = None
- self._on_unsubscribe: CallbackOnUnsubscribe | None = None
- self._on_disconnect: CallbackOnDisconnect | None = None
- self._on_socket_open: CallbackOnSocket | None = None
- self._on_socket_close: CallbackOnSocket | None = None
- self._on_socket_register_write: CallbackOnSocket | None = None
- self._on_socket_unregister_write: CallbackOnSocket | None = None
- self._websocket_path = "/mqtt"
- self._websocket_extra_headers: WebSocketHeaders | None = None
- # for clean_start == MQTT_CLEAN_START_FIRST_ONLY
- self._mqttv5_first_connect = True
- self.suppress_exceptions = False # For callbacks
-
- def __del__(self) -> None:
- self._reset_sockets()
-
- @property
- def host(self) -> str:
- """
- Host to connect to. If `connect()` hasn't been called yet, returns an empty string.
-
- This property may not be changed if the connection is already open.
- """
- return self._host
-
- @host.setter
- def host(self, value: str) -> None:
- if not self._connection_closed():
- raise RuntimeError("updating host on established connection is not supported")
-
- if not value:
- raise ValueError("Invalid host.")
- self._host = value
-
- @property
- def port(self) -> int:
- """
- Broker TCP port to connect to.
-
- This property may not be changed if the connection is already open.
- """
- return self._port
-
- @port.setter
- def port(self, value: int) -> None:
- if not self._connection_closed():
- raise RuntimeError("updating port on established connection is not supported")
-
- if value <= 0:
- raise ValueError("Invalid port number.")
- self._port = value
-
- @property
- def keepalive(self) -> int:
- """
- Client keepalive interval (in seconds).
-
- This property may not be changed if the connection is already open.
- """
- return self._keepalive
-
- @keepalive.setter
- def keepalive(self, value: int) -> None:
- if not self._connection_closed():
- # The issue here is that the previous value of keepalive matter to possibly
- # sent ping packet.
- raise RuntimeError("updating keepalive on established connection is not supported")
-
- if value < 0:
- raise ValueError("Keepalive must be >=0.")
-
- self._keepalive = value
-
- @property
- def transport(self) -> Literal["tcp", "websockets", "unix"]:
- """
- Transport method used for the connection ("tcp" or "websockets").
-
- This property may not be changed if the connection is already open.
- """
- return self._transport
-
- @transport.setter
- def transport(self, value: Literal["tcp", "websockets"]) -> None:
- if not self._connection_closed():
- raise RuntimeError("updating transport on established connection is not supported")
-
- self._transport = value
-
- @property
- def protocol(self) -> MQTTProtocolVersion:
- """
- Protocol version used (MQTT v3, MQTT v3.11, MQTTv5)
-
- This property is read-only.
- """
- return self._protocol
-
- @property
- def connect_timeout(self) -> float:
- """
- Connection establishment timeout in seconds.
-
- This property may not be changed if the connection is already open.
- """
- return self._connect_timeout
-
- @connect_timeout.setter
- def connect_timeout(self, value: float) -> None:
- if not self._connection_closed():
- raise RuntimeError("updating connect_timeout on established connection is not supported")
-
- if value <= 0.0:
- raise ValueError("timeout must be a positive number")
-
- self._connect_timeout = value
-
- @property
- def username(self) -> str | None:
- """The username used to connect to the MQTT broker, or None if no username is used.
-
- This property may not be changed if the connection is already open.
- """
- if self._username is None:
- return None
- return self._username.decode("utf-8")
-
- @username.setter
- def username(self, value: str | None) -> None:
- if not self._connection_closed():
- raise RuntimeError("updating username on established connection is not supported")
-
- if value is None:
- self._username = None
- else:
- self._username = value.encode("utf-8")
-
- @property
- def password(self) -> str | None:
- """The password used to connect to the MQTT broker, or None if no password is used.
-
- This property may not be changed if the connection is already open.
- """
- if self._password is None:
- return None
- return self._password.decode("utf-8")
-
- @password.setter
- def password(self, value: str | None) -> None:
- if not self._connection_closed():
- raise RuntimeError("updating password on established connection is not supported")
-
- if value is None:
- self._password = None
- else:
- self._password = value.encode("utf-8")
-
- @property
- def max_inflight_messages(self) -> int:
- """
- Maximum number of messages with QoS > 0 that can be partway through the network flow at once
-
- This property may not be changed if the connection is already open.
- """
- return self._max_inflight_messages
-
- @max_inflight_messages.setter
- def max_inflight_messages(self, value: int) -> None:
- if not self._connection_closed():
- # Not tested. Some doubt that everything is okay when max_inflight change between 0
- # and > 0 value because _update_inflight is skipped when _max_inflight_messages == 0
- raise RuntimeError("updating max_inflight_messages on established connection is not supported")
-
- if value < 0:
- raise ValueError("Invalid inflight.")
-
- self._max_inflight_messages = value
-
- @property
- def max_queued_messages(self) -> int:
- """
- Maximum number of message in the outgoing message queue, 0 means unlimited
-
- This property may not be changed if the connection is already open.
- """
- return self._max_queued_messages
-
- @max_queued_messages.setter
- def max_queued_messages(self, value: int) -> None:
- if not self._connection_closed():
- # Not tested.
- raise RuntimeError("updating max_queued_messages on established connection is not supported")
-
- if value < 0:
- raise ValueError("Invalid queue size.")
-
- self._max_queued_messages = value
-
- @property
- def will_topic(self) -> str | None:
- """
- The topic name a will message is sent to when disconnecting unexpectedly. None if a will shall not be sent.
-
- This property is read-only. Use `will_set()` to change its value.
- """
- if self._will_topic is None:
- return None
-
- return self._will_topic.decode("utf-8")
-
- @property
- def will_payload(self) -> bytes | None:
- """
- The payload for the will message that is sent when disconnecting unexpectedly. None if a will shall not be sent.
-
- This property is read-only. Use `will_set()` to change its value.
- """
- return self._will_payload
-
- @property
- def logger(self) -> logging.Logger | None:
- return self._logger
-
- @logger.setter
- def logger(self, value: logging.Logger | None) -> None:
- self._logger = value
-
- def _sock_recv(self, bufsize: int) -> bytes:
- if self._sock is None:
- raise ConnectionError("self._sock is None")
- try:
- return self._sock.recv(bufsize)
- except ssl.SSLWantReadError as err:
- raise BlockingIOError() from err
- except ssl.SSLWantWriteError as err:
- self._call_socket_register_write()
- raise BlockingIOError() from err
- except AttributeError as err:
- self._easy_log(
- MQTT_LOG_DEBUG, "socket was None: %s", err)
- raise ConnectionError() from err
-
- def _sock_send(self, buf: bytes) -> int:
- if self._sock is None:
- raise ConnectionError("self._sock is None")
-
- try:
- return self._sock.send(buf)
- except ssl.SSLWantReadError as err:
- raise BlockingIOError() from err
- except ssl.SSLWantWriteError as err:
- self._call_socket_register_write()
- raise BlockingIOError() from err
- except BlockingIOError as err:
- self._call_socket_register_write()
- raise BlockingIOError() from err
-
- def _sock_close(self) -> None:
- """Close the connection to the server."""
- if not self._sock:
- return
-
- try:
- sock = self._sock
- self._sock = None
- self._call_socket_unregister_write(sock)
- self._call_socket_close(sock)
- finally:
- # In case a callback fails, still close the socket to avoid leaking the file descriptor.
- sock.close()
-
- def _reset_sockets(self, sockpair_only: bool = False) -> None:
- if not sockpair_only:
- self._sock_close()
-
- if self._sockpairR:
- self._sockpairR.close()
- self._sockpairR = None
- if self._sockpairW:
- self._sockpairW.close()
- self._sockpairW = None
-
- def reinitialise(
- self,
- client_id: str = "",
- clean_session: bool = True,
- userdata: Any = None,
- ) -> None:
- self._reset_sockets()
-
- self.__init__(client_id, clean_session, userdata) # type: ignore[misc]
-
- def ws_set_options(
- self,
- path: str = "/mqtt",
- headers: WebSocketHeaders | None = None,
- ) -> None:
- """ Set the path and headers for a websocket connection
-
- :param str path: a string starting with / which should be the endpoint of the
- mqtt connection on the remote server
-
- :param headers: can be either a dict or a callable object. If it is a dict then
- the extra items in the dict are added to the websocket headers. If it is
- a callable, then the default websocket headers are passed into this
- function and the result is used as the new headers.
- """
- self._websocket_path = path
-
- if headers is not None:
- if isinstance(headers, dict) or callable(headers):
- self._websocket_extra_headers = headers
- else:
- raise ValueError(
- "'headers' option to ws_set_options has to be either a dictionary or callable")
-
- def tls_set_context(
- self,
- context: ssl.SSLContext | None = None,
- ) -> None:
- """Configure network encryption and authentication context. Enables SSL/TLS support.
-
- :param context: an ssl.SSLContext object. By default this is given by
- ``ssl.create_default_context()``, if available.
-
- Must be called before `connect()`, `connect_async()` or `connect_srv()`."""
- if self._ssl_context is not None:
- raise ValueError('SSL/TLS has already been configured.')
-
- if context is None:
- context = ssl.create_default_context()
-
- self._ssl = True
- self._ssl_context = context
-
- # Ensure _tls_insecure is consistent with check_hostname attribute
- if hasattr(context, 'check_hostname'):
- self._tls_insecure = not context.check_hostname
-
- def tls_set(
- self,
- ca_certs: str | None = None,
- certfile: str | None = None,
- keyfile: str | None = None,
- cert_reqs: ssl.VerifyMode | None = None,
- tls_version: int | None = None,
- ciphers: str | None = None,
- keyfile_password: str | None = None,
- alpn_protocols: list[str] | None = None,
- ) -> None:
- """Configure network encryption and authentication options. Enables SSL/TLS support.
-
- :param str ca_certs: a string path to the Certificate Authority certificate files
- that are to be treated as trusted by this client. If this is the only
- option given then the client will operate in a similar manner to a web
- browser. That is to say it will require the broker to have a
- certificate signed by the Certificate Authorities in ca_certs and will
- communicate using TLS v1,2, but will not attempt any form of
- authentication. This provides basic network encryption but may not be
- sufficient depending on how the broker is configured.
-
- By default, on Python 2.7.9+ or 3.4+, the default certification
- authority of the system is used. On older Python version this parameter
- is mandatory.
- :param str certfile: PEM encoded client certificate filename. Used with
- keyfile for client TLS based authentication. Support for this feature is
- broker dependent. Note that if the files in encrypted and needs a password to
- decrypt it, then this can be passed using the keyfile_password argument - you
- should take precautions to ensure that your password is
- not hard coded into your program by loading the password from a file
- for example. If you do not provide keyfile_password, the password will
- be requested to be typed in at a terminal window.
- :param str keyfile: PEM encoded client private keys filename. Used with
- certfile for client TLS based authentication. Support for this feature is
- broker dependent. Note that if the files in encrypted and needs a password to
- decrypt it, then this can be passed using the keyfile_password argument - you
- should take precautions to ensure that your password is
- not hard coded into your program by loading the password from a file
- for example. If you do not provide keyfile_password, the password will
- be requested to be typed in at a terminal window.
- :param cert_reqs: the certificate requirements that the client imposes
- on the broker to be changed. By default this is ssl.CERT_REQUIRED,
- which means that the broker must provide a certificate. See the ssl
- pydoc for more information on this parameter.
- :param tls_version: the version of the SSL/TLS protocol used to be
- specified. By default TLS v1.2 is used. Previous versions are allowed
- but not recommended due to possible security problems.
- :param str ciphers: encryption ciphers that are allowed
- for this connection, or None to use the defaults. See the ssl pydoc for
- more information.
-
- Must be called before `connect()`, `connect_async()` or `connect_srv()`."""
- if ssl is None:
- raise ValueError('This platform has no SSL/TLS.')
-
- if not hasattr(ssl, 'SSLContext'):
- # Require Python version that has SSL context support in standard library
- raise ValueError(
- 'Python 2.7.9 and 3.2 are the minimum supported versions for TLS.')
-
- if ca_certs is None and not hasattr(ssl.SSLContext, 'load_default_certs'):
- raise ValueError('ca_certs must not be None.')
-
- # Create SSLContext object
- if tls_version is None:
- tls_version = ssl.PROTOCOL_TLSv1_2
- # If the python version supports it, use highest TLS version automatically
- if hasattr(ssl, "PROTOCOL_TLS_CLIENT"):
- # This also enables CERT_REQUIRED and check_hostname by default.
- tls_version = ssl.PROTOCOL_TLS_CLIENT
- elif hasattr(ssl, "PROTOCOL_TLS"):
- tls_version = ssl.PROTOCOL_TLS
- context = ssl.SSLContext(tls_version)
-
- # Configure context
- if ciphers is not None:
- context.set_ciphers(ciphers)
-
- if certfile is not None:
- context.load_cert_chain(certfile, keyfile, keyfile_password)
-
- if cert_reqs == ssl.CERT_NONE and hasattr(context, 'check_hostname'):
- context.check_hostname = False
-
- context.verify_mode = ssl.CERT_REQUIRED if cert_reqs is None else cert_reqs
-
- if ca_certs is not None:
- context.load_verify_locations(ca_certs)
- else:
- context.load_default_certs()
-
- if alpn_protocols is not None:
- if not getattr(ssl, "HAS_ALPN", None):
- raise ValueError("SSL library has no support for ALPN")
- context.set_alpn_protocols(alpn_protocols)
-
- self.tls_set_context(context)
-
- if cert_reqs != ssl.CERT_NONE:
- # Default to secure, sets context.check_hostname attribute
- # if available
- self.tls_insecure_set(False)
- else:
- # But with ssl.CERT_NONE, we can not check_hostname
- self.tls_insecure_set(True)
-
- def tls_insecure_set(self, value: bool) -> None:
- """Configure verification of the server hostname in the server certificate.
-
- If value is set to true, it is impossible to guarantee that the host
- you are connecting to is not impersonating your server. This can be
- useful in initial server testing, but makes it possible for a malicious
- third party to impersonate your server through DNS spoofing, for
- example.
-
- Do not use this function in a real system. Setting value to true means
- there is no point using encryption.
-
- Must be called before `connect()` and after either `tls_set()` or
- `tls_set_context()`."""
-
- if self._ssl_context is None:
- raise ValueError(
- 'Must configure SSL context before using tls_insecure_set.')
-
- self._tls_insecure = value
-
- # Ensure check_hostname is consistent with _tls_insecure attribute
- if hasattr(self._ssl_context, 'check_hostname'):
- # Rely on SSLContext to check host name
- # If verify_mode is CERT_NONE then the host name will never be checked
- self._ssl_context.check_hostname = not value
-
- def proxy_set(self, **proxy_args: Any) -> None:
- """Configure proxying of MQTT connection. Enables support for SOCKS or
- HTTP proxies.
-
- Proxying is done through the PySocks library. Brief descriptions of the
- proxy_args parameters are below; see the PySocks docs for more info.
-
- (Required)
-
- :param proxy_type: One of {socks.HTTP, socks.SOCKS4, or socks.SOCKS5}
- :param proxy_addr: IP address or DNS name of proxy server
-
- (Optional)
-
- :param proxy_port: (int) port number of the proxy server. If not provided,
- the PySocks package default value will be utilized, which differs by proxy_type.
- :param proxy_rdns: boolean indicating whether proxy lookup should be performed
- remotely (True, default) or locally (False)
- :param proxy_username: username for SOCKS5 proxy, or userid for SOCKS4 proxy
- :param proxy_password: password for SOCKS5 proxy
-
- Example::
-
- mqttc.proxy_set(proxy_type=socks.HTTP, proxy_addr='1.2.3.4', proxy_port=4231)
- """
- if socks is None:
- raise ValueError("PySocks must be installed for proxy support.")
- elif not self._proxy_is_valid(proxy_args):
- raise ValueError("proxy_type and/or proxy_addr are invalid.")
- else:
- self._proxy = proxy_args
-
- def enable_logger(self, logger: logging.Logger | None = None) -> None:
- """
- Enables a logger to send log messages to
-
- :param logging.Logger logger: if specified, that ``logging.Logger`` object will be used, otherwise
- one will be created automatically.
-
- See `disable_logger` to undo this action.
- """
- if logger is None:
- if self._logger is not None:
- # Do not replace existing logger
- return
- logger = logging.getLogger(__name__)
- self.logger = logger
-
- def disable_logger(self) -> None:
- """
- Disable logging using standard python logging package. This has no effect on the `on_log` callback.
- """
- self._logger = None
-
- def connect(
- self,
- host: str,
- port: int = 1883,
- keepalive: int = 60,
- bind_address: str = "",
- bind_port: int = 0,
- clean_start: CleanStartOption = MQTT_CLEAN_START_FIRST_ONLY,
- properties: Properties | None = None,
- ) -> MQTTErrorCode:
- """Connect to a remote broker. This is a blocking call that establishes
- the underlying connection and transmits a CONNECT packet.
- Note that the connection status will not be updated until a CONNACK is received and
- processed (this requires a running network loop, see `loop_start`, `loop_forever`, `loop`...).
-
- :param str host: the hostname or IP address of the remote broker.
- :param int port: the network port of the server host to connect to. Defaults to
- 1883. Note that the default port for MQTT over SSL/TLS is 8883 so if you
- are using `tls_set()` the port may need providing.
- :param int keepalive: Maximum period in seconds between communications with the
- broker. If no other messages are being exchanged, this controls the
- rate at which the client will send ping messages to the broker.
- :param bool clean_start: (MQTT v5.0 only) True, False or MQTT_CLEAN_START_FIRST_ONLY.
- Sets the MQTT v5.0 clean_start flag always, never or on the first successful connect only,
- respectively. MQTT session data (such as outstanding messages and subscriptions)
- is cleared on successful connect when the clean_start flag is set.
- For MQTT v3.1.1, the ``clean_session`` argument of `Client` should be used for similar
- result.
- :param Properties properties: (MQTT v5.0 only) the MQTT v5.0 properties to be sent in the
- MQTT connect packet.
- """
-
- if self._protocol == MQTTv5:
- self._mqttv5_first_connect = True
- else:
- if clean_start != MQTT_CLEAN_START_FIRST_ONLY:
- raise ValueError("Clean start only applies to MQTT V5")
- if properties:
- raise ValueError("Properties only apply to MQTT V5")
-
- self.connect_async(host, port, keepalive,
- bind_address, bind_port, clean_start, properties)
- return self.reconnect()
-
- def connect_srv(
- self,
- domain: str | None = None,
- keepalive: int = 60,
- bind_address: str = "",
- bind_port: int = 0,
- clean_start: CleanStartOption = MQTT_CLEAN_START_FIRST_ONLY,
- properties: Properties | None = None,
- ) -> MQTTErrorCode:
- """Connect to a remote broker.
-
- :param str domain: the DNS domain to search for SRV records; if None,
- try to determine local domain name.
- :param keepalive, bind_address, clean_start and properties: see `connect()`
- """
-
- if HAVE_DNS is False:
- raise ValueError(
- 'No DNS resolver library found, try "pip install dnspython".')
-
- if domain is None:
- domain = socket.getfqdn()
- domain = domain[domain.find('.') + 1:]
-
- try:
- rr = f'_mqtt._tcp.{domain}'
- if self._ssl:
- # IANA specifies secure-mqtt (not mqtts) for port 8883
- rr = f'_secure-mqtt._tcp.{domain}'
- answers = []
- for answer in dns.resolver.query(rr, dns.rdatatype.SRV):
- addr = answer.target.to_text()[:-1]
- answers.append(
- (addr, answer.port, answer.priority, answer.weight))
- except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer, dns.resolver.NoNameservers) as err:
- raise ValueError(f"No answer/NXDOMAIN for SRV in {domain}") from err
-
- # FIXME: doesn't account for weight
- for answer in answers:
- host, port, prio, weight = answer
-
- try:
- return self.connect(host, port, keepalive, bind_address, bind_port, clean_start, properties)
- except Exception: # noqa: S110
- pass
-
- raise ValueError("No SRV hosts responded")
-
- def connect_async(
- self,
- host: str,
- port: int = 1883,
- keepalive: int = 60,
- bind_address: str = "",
- bind_port: int = 0,
- clean_start: CleanStartOption = MQTT_CLEAN_START_FIRST_ONLY,
- properties: Properties | None = None,
- ) -> None:
- """Connect to a remote broker asynchronously. This is a non-blocking
- connect call that can be used with `loop_start()` to provide very quick
- start.
-
- Any already established connection will be terminated immediately.
-
- :param str host: the hostname or IP address of the remote broker.
- :param int port: the network port of the server host to connect to. Defaults to
- 1883. Note that the default port for MQTT over SSL/TLS is 8883 so if you
- are using `tls_set()` the port may need providing.
- :param int keepalive: Maximum period in seconds between communications with the
- broker. If no other messages are being exchanged, this controls the
- rate at which the client will send ping messages to the broker.
- :param bool clean_start: (MQTT v5.0 only) True, False or MQTT_CLEAN_START_FIRST_ONLY.
- Sets the MQTT v5.0 clean_start flag always, never or on the first successful connect only,
- respectively. MQTT session data (such as outstanding messages and subscriptions)
- is cleared on successful connect when the clean_start flag is set.
- For MQTT v3.1.1, the ``clean_session`` argument of `Client` should be used for similar
- result.
- :param Properties properties: (MQTT v5.0 only) the MQTT v5.0 properties to be sent in the
- MQTT connect packet.
- """
- if bind_port < 0:
- raise ValueError('Invalid bind port number.')
-
- # Switch to state NEW to allow update of host, port & co.
- self._sock_close()
- self._state = _ConnectionState.MQTT_CS_NEW
-
- self.host = host
- self.port = port
- self.keepalive = keepalive
- self._bind_address = bind_address
- self._bind_port = bind_port
- self._clean_start = clean_start
- self._connect_properties = properties
- self._state = _ConnectionState.MQTT_CS_CONNECT_ASYNC
-
- def reconnect_delay_set(self, min_delay: int = 1, max_delay: int = 120) -> None:
- """ Configure the exponential reconnect delay
-
- When connection is lost, wait initially min_delay seconds and
- double this time every attempt. The wait is capped at max_delay.
- Once the client is fully connected (e.g. not only TCP socket, but
- received a success CONNACK), the wait timer is reset to min_delay.
- """
- with self._reconnect_delay_mutex:
- self._reconnect_min_delay = min_delay
- self._reconnect_max_delay = max_delay
- self._reconnect_delay = None
-
- def reconnect(self) -> MQTTErrorCode:
- """Reconnect the client after a disconnect. Can only be called after
- connect()/connect_async()."""
- if len(self._host) == 0:
- raise ValueError('Invalid host.')
- if self._port <= 0:
- raise ValueError('Invalid port number.')
-
- self._in_packet = {
- "command": 0,
- "have_remaining": 0,
- "remaining_count": [],
- "remaining_mult": 1,
- "remaining_length": 0,
- "packet": bytearray(b""),
- "to_process": 0,
- "pos": 0,
- }
-
- self._ping_t = 0.0
- self._state = _ConnectionState.MQTT_CS_CONNECTING
-
- self._sock_close()
-
- # Mark all currently outgoing QoS = 0 packets as lost,
- # or `wait_for_publish()` could hang forever
- for pkt in self._out_packet:
- if pkt["command"] & 0xF0 == PUBLISH and pkt["qos"] == 0 and pkt["info"] is not None:
- pkt["info"].rc = MQTT_ERR_CONN_LOST
- pkt["info"]._set_as_published()
-
- self._out_packet.clear()
-
- with self._msgtime_mutex:
- self._last_msg_in = time_func()
- self._last_msg_out = time_func()
-
- # Put messages in progress in a valid state.
- self._messages_reconnect_reset()
-
- with self._callback_mutex:
- on_pre_connect = self.on_pre_connect
-
- if on_pre_connect:
- try:
- on_pre_connect(self, self._userdata)
- except Exception as err:
- self._easy_log(
- MQTT_LOG_ERR, 'Caught exception in on_pre_connect: %s', err)
- if not self.suppress_exceptions:
- raise
-
- self._sock = self._create_socket()
-
- self._sock.setblocking(False) # type: ignore[attr-defined]
- self._registered_write = False
- self._call_socket_open(self._sock)
-
- return self._send_connect(self._keepalive)
-
- def loop(self, timeout: float = 1.0) -> MQTTErrorCode:
- """Process network events.
-
- It is strongly recommended that you use `loop_start()`, or
- `loop_forever()`, or if you are using an external event loop using
- `loop_read()`, `loop_write()`, and `loop_misc()`. Using loop() on it's own is
- no longer recommended.
-
- This function must be called regularly to ensure communication with the
- broker is carried out. It calls select() on the network socket to wait
- for network events. If incoming data is present it will then be
- processed. Outgoing commands, from e.g. `publish()`, are normally sent
- immediately that their function is called, but this is not always
- possible. loop() will also attempt to send any remaining outgoing
- messages, which also includes commands that are part of the flow for
- messages with QoS>0.
-
- :param int timeout: The time in seconds to wait for incoming/outgoing network
- traffic before timing out and returning.
-
- Returns MQTT_ERR_SUCCESS on success.
- Returns >0 on error.
-
- A ValueError will be raised if timeout < 0"""
-
- if self._sockpairR is None or self._sockpairW is None:
- self._reset_sockets(sockpair_only=True)
- self._sockpairR, self._sockpairW = _socketpair_compat()
-
- return self._loop(timeout)
-
- def _loop(self, timeout: float = 1.0) -> MQTTErrorCode:
- if timeout < 0.0:
- raise ValueError('Invalid timeout.')
-
- if self.want_write():
- wlist = [self._sock]
- else:
- wlist = []
-
- # used to check if there are any bytes left in the (SSL) socket
- pending_bytes = 0
- if hasattr(self._sock, 'pending'):
- pending_bytes = self._sock.pending() # type: ignore[union-attr]
-
- # if bytes are pending do not wait in select
- if pending_bytes > 0:
- timeout = 0.0
-
- # sockpairR is used to break out of select() before the timeout, on a
- # call to publish() etc.
- if self._sockpairR is None:
- rlist = [self._sock]
- else:
- rlist = [self._sock, self._sockpairR]
-
- try:
- socklist = select.select(rlist, wlist, [], timeout)
- except TypeError:
- # Socket isn't correct type, in likelihood connection is lost
- # ... or we called disconnect(). In that case the socket will
- # be closed but some loop (like loop_forever) will continue to
- # call _loop(). We still want to break that loop by returning an
- # rc != MQTT_ERR_SUCCESS and we don't want state to change from
- # mqtt_cs_disconnecting.
- if self._state not in (_ConnectionState.MQTT_CS_DISCONNECTING, _ConnectionState.MQTT_CS_DISCONNECTED):
- self._state = _ConnectionState.MQTT_CS_CONNECTION_LOST
- return MQTTErrorCode.MQTT_ERR_CONN_LOST
- except ValueError:
- # Can occur if we just reconnected but rlist/wlist contain a -1 for
- # some reason.
- if self._state not in (_ConnectionState.MQTT_CS_DISCONNECTING, _ConnectionState.MQTT_CS_DISCONNECTED):
- self._state = _ConnectionState.MQTT_CS_CONNECTION_LOST
- return MQTTErrorCode.MQTT_ERR_CONN_LOST
- except Exception:
- # Note that KeyboardInterrupt, etc. can still terminate since they
- # are not derived from Exception
- return MQTTErrorCode.MQTT_ERR_UNKNOWN
-
- if self._sock in socklist[0] or pending_bytes > 0:
- rc = self.loop_read()
- if rc or self._sock is None:
- return rc
-
- if self._sockpairR and self._sockpairR in socklist[0]:
- # Stimulate output write even though we didn't ask for it, because
- # at that point the publish or other command wasn't present.
- socklist[1].insert(0, self._sock)
- # Clear sockpairR - only ever a single byte written.
- try:
- # Read many bytes at once - this allows up to 10000 calls to
- # publish() inbetween calls to loop().
- self._sockpairR.recv(10000)
- except BlockingIOError:
- pass
-
- if self._sock in socklist[1]:
- rc = self.loop_write()
- if rc or self._sock is None:
- return rc
-
- return self.loop_misc()
-
- def publish(
- self,
- topic: str,
- payload: PayloadType = None,
- qos: int = 0,
- retain: bool = False,
- properties: Properties | None = None,
- ) -> MQTTMessageInfo:
- """Publish a message on a topic.
-
- This causes a message to be sent to the broker and subsequently from
- the broker to any clients subscribing to matching topics.
-
- :param str topic: The topic that the message should be published on.
- :param payload: The actual message to send. If not given, or set to None a
- zero length message will be used. Passing an int or float will result
- in the payload being converted to a string representing that number. If
- you wish to send a true int/float, use struct.pack() to create the
- payload you require.
- :param int qos: The quality of service level to use.
- :param bool retain: If set to true, the message will be set as the "last known
- good"/retained message for the topic.
- :param Properties properties: (MQTT v5.0 only) the MQTT v5.0 properties to be included.
-
- Returns a `MQTTMessageInfo` class, which can be used to determine whether
- the message has been delivered (using `is_published()`) or to block
- waiting for the message to be delivered (`wait_for_publish()`). The
- message ID and return code of the publish() call can be found at
- :py:attr:`info.mid <MQTTMessage.mid>` and :py:attr:`info.rc <MQTTMessage.rc>`.
-
- For backwards compatibility, the `MQTTMessageInfo` class is iterable so
- the old construct of ``(rc, mid) = client.publish(...)`` is still valid.
-
- rc is MQTT_ERR_SUCCESS to indicate success or MQTT_ERR_NO_CONN if the
- client is not currently connected. mid is the message ID for the
- publish request. The mid value can be used to track the publish request
- by checking against the mid argument in the on_publish() callback if it
- is defined.
-
- :raises ValueError: if topic is None, has zero length or is
- invalid (contains a wildcard), except if the MQTT version used is v5.0.
- For v5.0, a zero length topic can be used when a Topic Alias has been set.
- :raises ValueError: if qos is not one of 0, 1 or 2
- :raises ValueError: if the length of the payload is greater than 268435455 bytes.
- """
- if self._protocol != MQTTv5:
- if topic is None or len(topic) == 0:
- raise ValueError('Invalid topic.')
-
- topic_bytes = topic.encode('utf-8')
-
- self._raise_for_invalid_topic(topic_bytes)
-
- if qos < 0 or qos > 2:
- raise ValueError('Invalid QoS level.')
-
- local_payload = _encode_payload(payload)
-
- if len(local_payload) > 268435455:
- raise ValueError('Payload too large.')
-
- local_mid = self._mid_generate()
-
- if qos == 0:
- info = MQTTMessageInfo(local_mid)
- rc = self._send_publish(
- local_mid, topic_bytes, local_payload, qos, retain, False, info, properties)
- info.rc = rc
- return info
- else:
- message = MQTTMessage(local_mid, topic_bytes)
- message.timestamp = time_func()
- message.payload = local_payload
- message.qos = qos
- message.retain = retain
- message.dup = False
- message.properties = properties
-
- with self._out_message_mutex:
- if self._max_queued_messages > 0 and len(self._out_messages) >= self._max_queued_messages:
- message.info.rc = MQTTErrorCode.MQTT_ERR_QUEUE_SIZE
- return message.info
-
- if local_mid in self._out_messages:
- message.info.rc = MQTTErrorCode.MQTT_ERR_QUEUE_SIZE
- return message.info
-
- self._out_messages[message.mid] = message
- if self._max_inflight_messages == 0 or self._inflight_messages < self._max_inflight_messages:
- self._inflight_messages += 1
- if qos == 1:
- message.state = mqtt_ms_wait_for_puback
- elif qos == 2:
- message.state = mqtt_ms_wait_for_pubrec
-
- rc = self._send_publish(message.mid, topic_bytes, message.payload, message.qos, message.retain,
- message.dup, message.info, message.properties)
-
- # remove from inflight messages so it will be send after a connection is made
- if rc == MQTTErrorCode.MQTT_ERR_NO_CONN:
- self._inflight_messages -= 1
- message.state = mqtt_ms_publish
-
- message.info.rc = rc
- return message.info
- else:
- message.state = mqtt_ms_queued
- message.info.rc = MQTTErrorCode.MQTT_ERR_SUCCESS
- return message.info
-
- def username_pw_set(
- self, username: str | None, password: str | None = None
- ) -> None:
- """Set a username and optionally a password for broker authentication.
-
- Must be called before connect() to have any effect.
- Requires a broker that supports MQTT v3.1 or more.
-
- :param str username: The username to authenticate with. Need have no relationship to the client id. Must be str
- [MQTT-3.1.3-11].
- Set to None to reset client back to not using username/password for broker authentication.
- :param str password: The password to authenticate with. Optional, set to None if not required. If it is str, then it
- will be encoded as UTF-8.
- """
-
- # [MQTT-3.1.3-11] User name must be UTF-8 encoded string
- self._username = None if username is None else username.encode('utf-8')
- if isinstance(password, str):
- self._password = password.encode('utf-8')
- else:
- self._password = password
-
- def enable_bridge_mode(self) -> None:
- """Sets the client in a bridge mode instead of client mode.
-
- Must be called before `connect()` to have any effect.
- Requires brokers that support bridge mode.
-
- Under bridge mode, the broker will identify the client as a bridge and
- not send it's own messages back to it. Hence a subsciption of # is
- possible without message loops. This feature also correctly propagates
- the retain flag on the messages.
-
- Currently Mosquitto and RSMB support this feature. This feature can
- be used to create a bridge between multiple broker.
- """
- self._client_mode = MQTT_BRIDGE
-
- def _connection_closed(self) -> bool:
- """
- Return true if the connection is closed (and not trying to be opened).
- """
- return (
- self._state == _ConnectionState.MQTT_CS_NEW
- or (self._state in (_ConnectionState.MQTT_CS_DISCONNECTING, _ConnectionState.MQTT_CS_DISCONNECTED) and self._sock is None))
-
- def is_connected(self) -> bool:
- """Returns the current status of the connection
-
- True if connection exists
- False if connection is closed
- """
- return self._state == _ConnectionState.MQTT_CS_CONNECTED
-
- def disconnect(
- self,
- reasoncode: ReasonCode | None = None,
- properties: Properties | None = None,
- ) -> MQTTErrorCode:
- """Disconnect a connected client from the broker.
-
- :param ReasonCode reasoncode: (MQTT v5.0 only) a ReasonCode instance setting the MQTT v5.0
- reasoncode to be sent with the disconnect packet. It is optional, the receiver
- then assuming that 0 (success) is the value.
- :param Properties properties: (MQTT v5.0 only) a Properties instance setting the MQTT v5.0 properties
- to be included. Optional - if not set, no properties are sent.
- """
- if self._sock is None:
- self._state = _ConnectionState.MQTT_CS_DISCONNECTED
- return MQTT_ERR_NO_CONN
- else:
- self._state = _ConnectionState.MQTT_CS_DISCONNECTING
-
- return self._send_disconnect(reasoncode, properties)
-
- def subscribe(
- self,
- topic: str | tuple[str, int] | tuple[str, SubscribeOptions] | list[tuple[str, int]] | list[tuple[str, SubscribeOptions]],
- qos: int = 0,
- options: SubscribeOptions | None = None,
- properties: Properties | None = None,
- ) -> tuple[MQTTErrorCode, int | None]:
- """Subscribe the client to one or more topics.
-
- This function may be called in three different ways (and a further three for MQTT v5.0):
-
- Simple string and integer
- -------------------------
- e.g. subscribe("my/topic", 2)
-
- :topic: A string specifying the subscription topic to subscribe to.
- :qos: The desired quality of service level for the subscription.
- Defaults to 0.
- :options and properties: Not used.
-
- Simple string and subscribe options (MQTT v5.0 only)
- ----------------------------------------------------
- e.g. subscribe("my/topic", options=SubscribeOptions(qos=2))
-
- :topic: A string specifying the subscription topic to subscribe to.
- :qos: Not used.
- :options: The MQTT v5.0 subscribe options.
- :properties: a Properties instance setting the MQTT v5.0 properties
- to be included. Optional - if not set, no properties are sent.
-
- String and integer tuple
- ------------------------
- e.g. subscribe(("my/topic", 1))
-
- :topic: A tuple of (topic, qos). Both topic and qos must be present in
- the tuple.
- :qos and options: Not used.
- :properties: Only used for MQTT v5.0. A Properties instance setting the
- MQTT v5.0 properties. Optional - if not set, no properties are sent.
-
- String and subscribe options tuple (MQTT v5.0 only)
- ---------------------------------------------------
- e.g. subscribe(("my/topic", SubscribeOptions(qos=1)))
-
- :topic: A tuple of (topic, SubscribeOptions). Both topic and subscribe
- options must be present in the tuple.
- :qos and options: Not used.
- :properties: a Properties instance setting the MQTT v5.0 properties
- to be included. Optional - if not set, no properties are sent.
-
- List of string and integer tuples
- ---------------------------------
- e.g. subscribe([("my/topic", 0), ("another/topic", 2)])
-
- This allows multiple topic subscriptions in a single SUBSCRIPTION
- command, which is more efficient than using multiple calls to
- subscribe().
-
- :topic: A list of tuple of format (topic, qos). Both topic and qos must
- be present in all of the tuples.
- :qos, options and properties: Not used.
-
- List of string and subscribe option tuples (MQTT v5.0 only)
- -----------------------------------------------------------
- e.g. subscribe([("my/topic", SubscribeOptions(qos=0), ("another/topic", SubscribeOptions(qos=2)])
-
- This allows multiple topic subscriptions in a single SUBSCRIPTION
- command, which is more efficient than using multiple calls to
- subscribe().
-
- :topic: A list of tuple of format (topic, SubscribeOptions). Both topic and subscribe
- options must be present in all of the tuples.
- :qos and options: Not used.
- :properties: a Properties instance setting the MQTT v5.0 properties
- to be included. Optional - if not set, no properties are sent.
-
- The function returns a tuple (result, mid), where result is
- MQTT_ERR_SUCCESS to indicate success or (MQTT_ERR_NO_CONN, None) if the
- client is not currently connected. mid is the message ID for the
- subscribe request. The mid value can be used to track the subscribe
- request by checking against the mid argument in the on_subscribe()
- callback if it is defined.
-
- Raises a ValueError if qos is not 0, 1 or 2, or if topic is None or has
- zero string length, or if topic is not a string, tuple or list.
- """
- topic_qos_list = None
-
- if isinstance(topic, tuple):
- if self._protocol == MQTTv5:
- topic, options = topic # type: ignore
- if not isinstance(options, SubscribeOptions):
- raise ValueError(
- 'Subscribe options must be instance of SubscribeOptions class.')
- else:
- topic, qos = topic # type: ignore
-
- if isinstance(topic, (bytes, str)):
- if qos < 0 or qos > 2:
- raise ValueError('Invalid QoS level.')
- if self._protocol == MQTTv5:
- if options is None:
- # if no options are provided, use the QoS passed instead
- options = SubscribeOptions(qos=qos)
- elif qos != 0:
- raise ValueError(
- 'Subscribe options and qos parameters cannot be combined.')
- if not isinstance(options, SubscribeOptions):
- raise ValueError(
- 'Subscribe options must be instance of SubscribeOptions class.')
- topic_qos_list = [(topic.encode('utf-8'), options)]
- else:
- if topic is None or len(topic) == 0:
- raise ValueError('Invalid topic.')
- topic_qos_list = [(topic.encode('utf-8'), qos)] # type: ignore
- elif isinstance(topic, list):
- if len(topic) == 0:
- raise ValueError('Empty topic list')
- topic_qos_list = []
- if self._protocol == MQTTv5:
- for t, o in topic:
- if not isinstance(o, SubscribeOptions):
- # then the second value should be QoS
- if o < 0 or o > 2:
- raise ValueError('Invalid QoS level.')
- o = SubscribeOptions(qos=o)
- topic_qos_list.append((t.encode('utf-8'), o))
- else:
- for t, q in topic:
- if isinstance(q, SubscribeOptions) or q < 0 or q > 2:
- raise ValueError('Invalid QoS level.')
- if t is None or len(t) == 0 or not isinstance(t, (bytes, str)):
- raise ValueError('Invalid topic.')
- topic_qos_list.append((t.encode('utf-8'), q)) # type: ignore
-
- if topic_qos_list is None:
- raise ValueError("No topic specified, or incorrect topic type.")
-
- if any(self._filter_wildcard_len_check(topic) != MQTT_ERR_SUCCESS for topic, _ in topic_qos_list):
- raise ValueError('Invalid subscription filter.')
-
- if self._sock is None:
- return (MQTT_ERR_NO_CONN, None)
-
- return self._send_subscribe(False, topic_qos_list, properties)
-
- def unsubscribe(
- self, topic: str | list[str], properties: Properties | None = None
- ) -> tuple[MQTTErrorCode, int | None]:
- """Unsubscribe the client from one or more topics.
-
- :param topic: A single string, or list of strings that are the subscription
- topics to unsubscribe from.
- :param properties: (MQTT v5.0 only) a Properties instance setting the MQTT v5.0 properties
- to be included. Optional - if not set, no properties are sent.
-
- Returns a tuple (result, mid), where result is MQTT_ERR_SUCCESS
- to indicate success or (MQTT_ERR_NO_CONN, None) if the client is not
- currently connected.
- mid is the message ID for the unsubscribe request. The mid value can be
- used to track the unsubscribe request by checking against the mid
- argument in the on_unsubscribe() callback if it is defined.
-
- :raises ValueError: if topic is None or has zero string length, or is
- not a string or list.
- """
- topic_list = None
- if topic is None:
- raise ValueError('Invalid topic.')
- if isinstance(topic, (bytes, str)):
- if len(topic) == 0:
- raise ValueError('Invalid topic.')
- topic_list = [topic.encode('utf-8')]
- elif isinstance(topic, list):
- topic_list = []
- for t in topic:
- if len(t) == 0 or not isinstance(t, (bytes, str)):
- raise ValueError('Invalid topic.')
- topic_list.append(t.encode('utf-8'))
-
- if topic_list is None:
- raise ValueError("No topic specified, or incorrect topic type.")
-
- if self._sock is None:
- return (MQTTErrorCode.MQTT_ERR_NO_CONN, None)
-
- return self._send_unsubscribe(False, topic_list, properties)
-
- def loop_read(self, max_packets: int = 1) -> MQTTErrorCode:
- """Process read network events. Use in place of calling `loop()` if you
- wish to handle your client reads as part of your own application.
-
- Use `socket()` to obtain the client socket to call select() or equivalent
- on.
-
- Do not use if you are using `loop_start()` or `loop_forever()`."""
- if self._sock is None:
- return MQTTErrorCode.MQTT_ERR_NO_CONN
-
- max_packets = len(self._out_messages) + len(self._in_messages)
- if max_packets < 1:
- max_packets = 1
-
- for _ in range(0, max_packets):
- if self._sock is None:
- return MQTTErrorCode.MQTT_ERR_NO_CONN
- rc = self._packet_read()
- if rc > 0:
- return self._loop_rc_handle(rc)
- elif rc == MQTTErrorCode.MQTT_ERR_AGAIN:
- return MQTTErrorCode.MQTT_ERR_SUCCESS
- return MQTTErrorCode.MQTT_ERR_SUCCESS
-
- def loop_write(self) -> MQTTErrorCode:
- """Process write network events. Use in place of calling `loop()` if you
- wish to handle your client writes as part of your own application.
-
- Use `socket()` to obtain the client socket to call select() or equivalent
- on.
-
- Use `want_write()` to determine if there is data waiting to be written.
-
- Do not use if you are using `loop_start()` or `loop_forever()`."""
- if self._sock is None:
- return MQTTErrorCode.MQTT_ERR_NO_CONN
-
- try:
- rc = self._packet_write()
- if rc == MQTTErrorCode.MQTT_ERR_AGAIN:
- return MQTTErrorCode.MQTT_ERR_SUCCESS
- elif rc > 0:
- return self._loop_rc_handle(rc)
- else:
- return MQTTErrorCode.MQTT_ERR_SUCCESS
- finally:
- if self.want_write():
- self._call_socket_register_write()
- else:
- self._call_socket_unregister_write()
-
- def want_write(self) -> bool:
- """Call to determine if there is network data waiting to be written.
- Useful if you are calling select() yourself rather than using `loop()`, `loop_start()` or `loop_forever()`.
- """
- return len(self._out_packet) > 0
-
- def loop_misc(self) -> MQTTErrorCode:
- """Process miscellaneous network events. Use in place of calling `loop()` if you
- wish to call select() or equivalent on.
-
- Do not use if you are using `loop_start()` or `loop_forever()`."""
- if self._sock is None:
- return MQTTErrorCode.MQTT_ERR_NO_CONN
-
- now = time_func()
- self._check_keepalive()
-
- if self._ping_t > 0 and now - self._ping_t >= self._keepalive:
- # client->ping_t != 0 means we are waiting for a pingresp.
- # This hasn't happened in the keepalive time so we should disconnect.
- self._sock_close()
-
- if self._state in (_ConnectionState.MQTT_CS_DISCONNECTING, _ConnectionState.MQTT_CS_DISCONNECTED):
- self._state = _ConnectionState.MQTT_CS_DISCONNECTED
- rc = MQTTErrorCode.MQTT_ERR_SUCCESS
- else:
- self._state = _ConnectionState.MQTT_CS_CONNECTION_LOST
- rc = MQTTErrorCode.MQTT_ERR_KEEPALIVE
-
- self._do_on_disconnect(
- packet_from_broker=False,
- v1_rc=rc,
- )
-
- return MQTTErrorCode.MQTT_ERR_CONN_LOST
-
- return MQTTErrorCode.MQTT_ERR_SUCCESS
-
- def max_inflight_messages_set(self, inflight: int) -> None:
- """Set the maximum number of messages with QoS>0 that can be part way
- through their network flow at once. Defaults to 20."""
- self.max_inflight_messages = inflight
-
- def max_queued_messages_set(self, queue_size: int) -> Client:
- """Set the maximum number of messages in the outgoing message queue.
- 0 means unlimited."""
- if not isinstance(queue_size, int):
- raise ValueError('Invalid type of queue size.')
- self.max_queued_messages = queue_size
- return self
-
- def user_data_set(self, userdata: Any) -> None:
- """Set the user data variable passed to callbacks. May be any data type."""
- self._userdata = userdata
-
- def user_data_get(self) -> Any:
- """Get the user data variable passed to callbacks. May be any data type."""
- return self._userdata
-
- def will_set(
- self,
- topic: str,
- payload: PayloadType = None,
- qos: int = 0,
- retain: bool = False,
- properties: Properties | None = None,
- ) -> None:
- """Set a Will to be sent by the broker in case the client disconnects unexpectedly.
-
- This must be called before connect() to have any effect.
-
- :param str topic: The topic that the will message should be published on.
- :param payload: The message to send as a will. If not given, or set to None a
- zero length message will be used as the will. Passing an int or float
- will result in the payload being converted to a string representing
- that number. If you wish to send a true int/float, use struct.pack() to
- create the payload you require.
- :param int qos: The quality of service level to use for the will.
- :param bool retain: If set to true, the will message will be set as the "last known
- good"/retained message for the topic.
- :param Properties properties: (MQTT v5.0 only) the MQTT v5.0 properties
- to be included with the will message. Optional - if not set, no properties are sent.
-
- :raises ValueError: if qos is not 0, 1 or 2, or if topic is None or has
- zero string length.
-
- See `will_clear` to clear will. Note that will are NOT send if the client disconnect cleanly
- for example by calling `disconnect()`.
- """
- if topic is None or len(topic) == 0:
- raise ValueError('Invalid topic.')
-
- if qos < 0 or qos > 2:
- raise ValueError('Invalid QoS level.')
-
- if properties and not isinstance(properties, Properties):
- raise ValueError(
- "The properties argument must be an instance of the Properties class.")
-
- self._will_payload = _encode_payload(payload)
- self._will = True
- self._will_topic = topic.encode('utf-8')
- self._will_qos = qos
- self._will_retain = retain
- self._will_properties = properties
-
- def will_clear(self) -> None:
- """ Removes a will that was previously configured with `will_set()`.
-
- Must be called before connect() to have any effect."""
- self._will = False
- self._will_topic = b""
- self._will_payload = b""
- self._will_qos = 0
- self._will_retain = False
-
- def socket(self) -> SocketLike | None:
- """Return the socket or ssl object for this client."""
- return self._sock
-
- def loop_forever(
- self,
- timeout: float = 1.0,
- retry_first_connection: bool = False,
- ) -> MQTTErrorCode:
- """This function calls the network loop functions for you in an
- infinite blocking loop. It is useful for the case where you only want
- to run the MQTT client loop in your program.
-
- loop_forever() will handle reconnecting for you if reconnect_on_failure is
- true (this is the default behavior). If you call `disconnect()` in a callback
- it will return.
-
- :param int timeout: The time in seconds to wait for incoming/outgoing network
- traffic before timing out and returning.
- :param bool retry_first_connection: Should the first connection attempt be retried on failure.
- This is independent of the reconnect_on_failure setting.
-
- :raises OSError: if the first connection fail unless retry_first_connection=True
- """
-
- run = True
-
- while run:
- if self._thread_terminate is True:
- break
-
- if self._state == _ConnectionState.MQTT_CS_CONNECT_ASYNC:
- try:
- self.reconnect()
- except OSError:
- self._handle_on_connect_fail()
- if not retry_first_connection:
- raise
- self._easy_log(
- MQTT_LOG_DEBUG, "Connection failed, retrying")
- self._reconnect_wait()
- else:
- break
-
- while run:
- rc = MQTTErrorCode.MQTT_ERR_SUCCESS
- while rc == MQTTErrorCode.MQTT_ERR_SUCCESS:
- rc = self._loop(timeout)
- # We don't need to worry about locking here, because we've
- # either called loop_forever() when in single threaded mode, or
- # in multi threaded mode when loop_stop() has been called and
- # so no other threads can access _out_packet or _messages.
- if (self._thread_terminate is True
- and len(self._out_packet) == 0
- and len(self._out_messages) == 0):
- rc = MQTTErrorCode.MQTT_ERR_NOMEM
- run = False
-
- def should_exit() -> bool:
- return (
- self._state in (_ConnectionState.MQTT_CS_DISCONNECTING, _ConnectionState.MQTT_CS_DISCONNECTED) or
- run is False or # noqa: B023 (uses the run variable from the outer scope on purpose)
- self._thread_terminate is True
- )
-
- if should_exit() or not self._reconnect_on_failure:
- run = False
- else:
- self._reconnect_wait()
-
- if should_exit():
- run = False
- else:
- try:
- self.reconnect()
- except OSError:
- self._handle_on_connect_fail()
- self._easy_log(
- MQTT_LOG_DEBUG, "Connection failed, retrying")
-
- return rc
-
- def loop_start(self) -> MQTTErrorCode:
- """This is part of the threaded client interface. Call this once to
- start a new thread to process network traffic. This provides an
- alternative to repeatedly calling `loop()` yourself.
-
- Under the hood, this will call `loop_forever` in a thread, which means that
- the thread will terminate if you call `disconnect()`
- """
- if self._thread is not None:
- return MQTTErrorCode.MQTT_ERR_INVAL
-
- self._sockpairR, self._sockpairW = _socketpair_compat()
- self._thread_terminate = False
- self._thread = threading.Thread(target=self._thread_main, name=f"paho-mqtt-client-{self._client_id.decode()}")
- self._thread.daemon = True
- self._thread.start()
-
- return MQTTErrorCode.MQTT_ERR_SUCCESS
-
- def loop_stop(self) -> MQTTErrorCode:
- """This is part of the threaded client interface. Call this once to
- stop the network thread previously created with `loop_start()`. This call
- will block until the network thread finishes.
-
- This don't guarantee that publish packet are sent, use `wait_for_publish` or
- `on_publish` to ensure `publish` are sent.
- """
- if self._thread is None:
- return MQTTErrorCode.MQTT_ERR_INVAL
-
- self._thread_terminate = True
- if threading.current_thread() != self._thread:
- self._thread.join()
-
- return MQTTErrorCode.MQTT_ERR_SUCCESS
-
- @property
- def callback_api_version(self) -> CallbackAPIVersion:
- """
- Return the callback API version used for user-callback. See docstring for
- each user-callback (`on_connect`, `on_publish`, ...) for details.
-
- This property is read-only.
- """
- return self._callback_api_version
-
- @property
- def on_log(self) -> CallbackOnLog | None:
- """The callback called when the client has log information.
- Defined to allow debugging.
-
- Expected signature is::
-
- log_callback(client, userdata, level, buf)
-
- :param Client client: the client instance for this callback
- :param userdata: the private user data as set in Client() or user_data_set()
- :param int level: gives the severity of the message and will be one of
- MQTT_LOG_INFO, MQTT_LOG_NOTICE, MQTT_LOG_WARNING,
- MQTT_LOG_ERR, and MQTT_LOG_DEBUG.
- :param str buf: the message itself
-
- Decorator: @client.log_callback() (``client`` is the name of the
- instance which this callback is being attached to)
- """
- return self._on_log
-
- @on_log.setter
- def on_log(self, func: CallbackOnLog | None) -> None:
- self._on_log = func
-
- def log_callback(self) -> Callable[[CallbackOnLog], CallbackOnLog]:
- def decorator(func: CallbackOnLog) -> CallbackOnLog:
- self.on_log = func
- return func
- return decorator
-
- @property
- def on_pre_connect(self) -> CallbackOnPreConnect | None:
- """The callback called immediately prior to the connection is made
- request.
-
- Expected signature (for all callback API version)::
-
- connect_callback(client, userdata)
-
- :parama Client client: the client instance for this callback
- :parama userdata: the private user data as set in Client() or user_data_set()
-
- Decorator: @client.pre_connect_callback() (``client`` is the name of the
- instance which this callback is being attached to)
-
- """
- return self._on_pre_connect
-
- @on_pre_connect.setter
- def on_pre_connect(self, func: CallbackOnPreConnect | None) -> None:
- with self._callback_mutex:
- self._on_pre_connect = func
-
- def pre_connect_callback(
- self,
- ) -> Callable[[CallbackOnPreConnect], CallbackOnPreConnect]:
- def decorator(func: CallbackOnPreConnect) -> CallbackOnPreConnect:
- self.on_pre_connect = func
- return func
- return decorator
-
- @property
- def on_connect(self) -> CallbackOnConnect | None:
- """The callback called when the broker reponds to our connection request.
-
- Expected signature for callback API version 2::
-
- connect_callback(client, userdata, connect_flags, reason_code, properties)
-
- Expected signature for callback API version 1 change with MQTT protocol version:
- * For MQTT v3.1 and v3.1.1 it's::
-
- connect_callback(client, userdata, flags, rc)
-
- * For MQTT v5.0 it's::
-
- connect_callback(client, userdata, flags, reason_code, properties)
-
-
- :param Client client: the client instance for this callback
- :param userdata: the private user data as set in Client() or user_data_set()
- :param ConnectFlags connect_flags: the flags for this connection
- :param ReasonCode reason_code: the connection reason code received from the broken.
- In MQTT v5.0 it's the reason code defined by the standard.
- In MQTT v3, we convert return code to a reason code, see
- `convert_connack_rc_to_reason_code()`.
- `ReasonCode` may be compared to integer.
- :param Properties properties: the MQTT v5.0 properties received from the broker.
- For MQTT v3.1 and v3.1.1 properties is not provided and an empty Properties
- object is always used.
- :param dict flags: response flags sent by the broker
- :param int rc: the connection result, should have a value of `ConnackCode`
-
- flags is a dict that contains response flags from the broker:
- flags['session present'] - this flag is useful for clients that are
- using clean session set to 0 only. If a client with clean
- session=0, that reconnects to a broker that it has previously
- connected to, this flag indicates whether the broker still has the
- session information for the client. If 1, the session still exists.
-
- The value of rc indicates success or not:
- - 0: Connection successful
- - 1: Connection refused - incorrect protocol version
- - 2: Connection refused - invalid client identifier
- - 3: Connection refused - server unavailable
- - 4: Connection refused - bad username or password
- - 5: Connection refused - not authorised
- - 6-255: Currently unused.
-
- Decorator: @client.connect_callback() (``client`` is the name of the
- instance which this callback is being attached to)
- """
- return self._on_connect
-
- @on_connect.setter
- def on_connect(self, func: CallbackOnConnect | None) -> None:
- with self._callback_mutex:
- self._on_connect = func
-
- def connect_callback(
- self,
- ) -> Callable[[CallbackOnConnect], CallbackOnConnect]:
- def decorator(func: CallbackOnConnect) -> CallbackOnConnect:
- self.on_connect = func
- return func
- return decorator
-
- @property
- def on_connect_fail(self) -> CallbackOnConnectFail | None:
- """The callback called when the client failed to connect
- to the broker.
-
- Expected signature is (for all callback_api_version)::
-
- connect_fail_callback(client, userdata)
-
- :param Client client: the client instance for this callback
- :parama userdata: the private user data as set in Client() or user_data_set()
-
- Decorator: @client.connect_fail_callback() (``client`` is the name of the
- instance which this callback is being attached to)
- """
- return self._on_connect_fail
-
- @on_connect_fail.setter
- def on_connect_fail(self, func: CallbackOnConnectFail | None) -> None:
- with self._callback_mutex:
- self._on_connect_fail = func
-
- def connect_fail_callback(
- self,
- ) -> Callable[[CallbackOnConnectFail], CallbackOnConnectFail]:
- def decorator(func: CallbackOnConnectFail) -> CallbackOnConnectFail:
- self.on_connect_fail = func
- return func
- return decorator
-
- @property
- def on_subscribe(self) -> CallbackOnSubscribe | None:
- """The callback called when the broker responds to a subscribe
- request.
-
- Expected signature for callback API version 2::
-
- subscribe_callback(client, userdata, mid, reason_code_list, properties)
-
- Expected signature for callback API version 1 change with MQTT protocol version:
- * For MQTT v3.1 and v3.1.1 it's::
-
- subscribe_callback(client, userdata, mid, granted_qos)
-
- * For MQTT v5.0 it's::
-
- subscribe_callback(client, userdata, mid, reason_code_list, properties)
-
- :param Client client: the client instance for this callback
- :param userdata: the private user data as set in Client() or user_data_set()
- :param int mid: matches the mid variable returned from the corresponding
- subscribe() call.
- :param list[ReasonCode] reason_code_list: reason codes received from the broker for each subscription.
- In MQTT v5.0 it's the reason code defined by the standard.
- In MQTT v3, we convert granted QoS to a reason code.
- It's a list of ReasonCode instances.
- :param Properties properties: the MQTT v5.0 properties received from the broker.
- For MQTT v3.1 and v3.1.1 properties is not provided and an empty Properties
- object is always used.
- :param list[int] granted_qos: list of integers that give the QoS level the broker has
- granted for each of the different subscription requests.
-
- Decorator: @client.subscribe_callback() (``client`` is the name of the
- instance which this callback is being attached to)
- """
- return self._on_subscribe
-
- @on_subscribe.setter
- def on_subscribe(self, func: CallbackOnSubscribe | None) -> None:
- with self._callback_mutex:
- self._on_subscribe = func
-
- def subscribe_callback(
- self,
- ) -> Callable[[CallbackOnSubscribe], CallbackOnSubscribe]:
- def decorator(func: CallbackOnSubscribe) -> CallbackOnSubscribe:
- self.on_subscribe = func
- return func
- return decorator
-
- @property
- def on_message(self) -> CallbackOnMessage | None:
- """The callback called when a message has been received on a topic
- that the client subscribes to.
-
- This callback will be called for every message received unless a
- `message_callback_add()` matched the message.
-
- Expected signature is (for all callback API version):
- message_callback(client, userdata, message)
-
- :param Client client: the client instance for this callback
- :param userdata: the private user data as set in Client() or user_data_set()
- :param MQTTMessage message: the received message.
- This is a class with members topic, payload, qos, retain.
-
- Decorator: @client.message_callback() (``client`` is the name of the
- instance which this callback is being attached to)
- """
- return self._on_message
-
- @on_message.setter
- def on_message(self, func: CallbackOnMessage | None) -> None:
- with self._callback_mutex:
- self._on_message = func
-
- def message_callback(
- self,
- ) -> Callable[[CallbackOnMessage], CallbackOnMessage]:
- def decorator(func: CallbackOnMessage) -> CallbackOnMessage:
- self.on_message = func
- return func
- return decorator
-
- @property
- def on_publish(self) -> CallbackOnPublish | None:
- """The callback called when a message that was to be sent using the
- `publish()` call has completed transmission to the broker.
-
- For messages with QoS levels 1 and 2, this means that the appropriate
- handshakes have completed. For QoS 0, this simply means that the message
- has left the client.
- This callback is important because even if the `publish()` call returns
- success, it does not always mean that the message has been sent.
-
- See also `wait_for_publish` which could be simpler to use.
-
- Expected signature for callback API version 2::
-
- publish_callback(client, userdata, mid, reason_code, properties)
-
- Expected signature for callback API version 1::
-
- publish_callback(client, userdata, mid)
-
- :param Client client: the client instance for this callback
- :param userdata: the private user data as set in Client() or user_data_set()
- :param int mid: matches the mid variable returned from the corresponding
- `publish()` call, to allow outgoing messages to be tracked.
- :param ReasonCode reason_code: the connection reason code received from the broken.
- In MQTT v5.0 it's the reason code defined by the standard.
- In MQTT v3 it's always the reason code Success
- :parama Properties properties: the MQTT v5.0 properties received from the broker.
- For MQTT v3.1 and v3.1.1 properties is not provided and an empty Properties
- object is always used.
-
- Note: for QoS = 0, the reason_code and the properties don't really exist, it's the client
- library that generate them. It's always an empty properties and a success reason code.
- Because the (MQTTv5) standard don't have reason code for PUBLISH packet, the library create them
- at PUBACK packet, as if the message was sent with QoS = 1.
-
- Decorator: @client.publish_callback() (``client`` is the name of the
- instance which this callback is being attached to)
-
- """
- return self._on_publish
-
- @on_publish.setter
- def on_publish(self, func: CallbackOnPublish | None) -> None:
- with self._callback_mutex:
- self._on_publish = func
-
- def publish_callback(
- self,
- ) -> Callable[[CallbackOnPublish], CallbackOnPublish]:
- def decorator(func: CallbackOnPublish) -> CallbackOnPublish:
- self.on_publish = func
- return func
- return decorator
-
- @property
- def on_unsubscribe(self) -> CallbackOnUnsubscribe | None:
- """The callback called when the broker responds to an unsubscribe
- request.
-
- Expected signature for callback API version 2::
-
- unsubscribe_callback(client, userdata, mid, reason_code_list, properties)
-
- Expected signature for callback API version 1 change with MQTT protocol version:
- * For MQTT v3.1 and v3.1.1 it's::
-
- unsubscribe_callback(client, userdata, mid)
-
- * For MQTT v5.0 it's::
-
- unsubscribe_callback(client, userdata, mid, properties, v1_reason_codes)
-
- :param Client client: the client instance for this callback
- :param userdata: the private user data as set in Client() or user_data_set()
- :param mid: matches the mid variable returned from the corresponding
- unsubscribe() call.
- :param list[ReasonCode] reason_code_list: reason codes received from the broker for each unsubscription.
- In MQTT v5.0 it's the reason code defined by the standard.
- In MQTT v3, there is not equivalent from broken and empty list
- is always used.
- :param Properties properties: the MQTT v5.0 properties received from the broker.
- For MQTT v3.1 and v3.1.1 properties is not provided and an empty Properties
- object is always used.
- :param v1_reason_codes: the MQTT v5.0 reason codes received from the broker for each
- unsubscribe topic. A list of ReasonCode instances OR a single
- ReasonCode when we unsubscribe from a single topic.
-
- Decorator: @client.unsubscribe_callback() (``client`` is the name of the
- instance which this callback is being attached to)
- """
- return self._on_unsubscribe
-
- @on_unsubscribe.setter
- def on_unsubscribe(self, func: CallbackOnUnsubscribe | None) -> None:
- with self._callback_mutex:
- self._on_unsubscribe = func
-
- def unsubscribe_callback(
- self,
- ) -> Callable[[CallbackOnUnsubscribe], CallbackOnUnsubscribe]:
- def decorator(func: CallbackOnUnsubscribe) -> CallbackOnUnsubscribe:
- self.on_unsubscribe = func
- return func
- return decorator
-
- @property
- def on_disconnect(self) -> CallbackOnDisconnect | None:
- """The callback called when the client disconnects from the broker.
-
- Expected signature for callback API version 2::
-
- disconnect_callback(client, userdata, disconnect_flags, reason_code, properties)
-
- Expected signature for callback API version 1 change with MQTT protocol version:
- * For MQTT v3.1 and v3.1.1 it's::
-
- disconnect_callback(client, userdata, rc)
-
- * For MQTT v5.0 it's::
-
- disconnect_callback(client, userdata, reason_code, properties)
-
- :param Client client: the client instance for this callback
- :param userdata: the private user data as set in Client() or user_data_set()
- :param DisconnectFlag disconnect_flags: the flags for this disconnection.
- :param ReasonCode reason_code: the disconnection reason code possibly received from the broker (see disconnect_flags).
- In MQTT v5.0 it's the reason code defined by the standard.
- In MQTT v3 it's never received from the broker, we convert an MQTTErrorCode,
- see `convert_disconnect_error_code_to_reason_code()`.
- `ReasonCode` may be compared to integer.
- :param Properties properties: the MQTT v5.0 properties received from the broker.
- For MQTT v3.1 and v3.1.1 properties is not provided and an empty Properties
- object is always used.
- :param int rc: the disconnection result
- The rc parameter indicates the disconnection state. If
- MQTT_ERR_SUCCESS (0), the callback was called in response to
- a disconnect() call. If any other value the disconnection
- was unexpected, such as might be caused by a network error.
-
- Decorator: @client.disconnect_callback() (``client`` is the name of the
- instance which this callback is being attached to)
-
- """
- return self._on_disconnect
-
- @on_disconnect.setter
- def on_disconnect(self, func: CallbackOnDisconnect | None) -> None:
- with self._callback_mutex:
- self._on_disconnect = func
-
- def disconnect_callback(
- self,
- ) -> Callable[[CallbackOnDisconnect], CallbackOnDisconnect]:
- def decorator(func: CallbackOnDisconnect) -> CallbackOnDisconnect:
- self.on_disconnect = func
- return func
- return decorator
-
- @property
- def on_socket_open(self) -> CallbackOnSocket | None:
- """The callback called just after the socket was opend.
-
- This should be used to register the socket to an external event loop for reading.
-
- Expected signature is (for all callback API version)::
-
- socket_open_callback(client, userdata, socket)
-
- :param Client client: the client instance for this callback
- :param userdata: the private user data as set in Client() or user_data_set()
- :param SocketLike sock: the socket which was just opened.
-
- Decorator: @client.socket_open_callback() (``client`` is the name of the
- instance which this callback is being attached to)
- """
- return self._on_socket_open
-
- @on_socket_open.setter
- def on_socket_open(self, func: CallbackOnSocket | None) -> None:
- with self._callback_mutex:
- self._on_socket_open = func
-
- def socket_open_callback(
- self,
- ) -> Callable[[CallbackOnSocket], CallbackOnSocket]:
- def decorator(func: CallbackOnSocket) -> CallbackOnSocket:
- self.on_socket_open = func
- return func
- return decorator
-
- def _call_socket_open(self, sock: SocketLike) -> None:
- """Call the socket_open callback with the just-opened socket"""
- with self._callback_mutex:
- on_socket_open = self.on_socket_open
-
- if on_socket_open:
- with self._in_callback_mutex:
- try:
- on_socket_open(self, self._userdata, sock)
- except Exception as err:
- self._easy_log(
- MQTT_LOG_ERR, 'Caught exception in on_socket_open: %s', err)
- if not self.suppress_exceptions:
- raise
-
- @property
- def on_socket_close(self) -> CallbackOnSocket | None:
- """The callback called just before the socket is closed.
-
- This should be used to unregister the socket from an external event loop for reading.
-
- Expected signature is (for all callback API version)::
-
- socket_close_callback(client, userdata, socket)
-
- :param Client client: the client instance for this callback
- :param userdata: the private user data as set in Client() or user_data_set()
- :param SocketLike sock: the socket which is about to be closed.
-
- Decorator: @client.socket_close_callback() (``client`` is the name of the
- instance which this callback is being attached to)
- """
- return self._on_socket_close
-
- @on_socket_close.setter
- def on_socket_close(self, func: CallbackOnSocket | None) -> None:
- with self._callback_mutex:
- self._on_socket_close = func
-
- def socket_close_callback(
- self,
- ) -> Callable[[CallbackOnSocket], CallbackOnSocket]:
- def decorator(func: CallbackOnSocket) -> CallbackOnSocket:
- self.on_socket_close = func
- return func
- return decorator
-
- def _call_socket_close(self, sock: SocketLike) -> None:
- """Call the socket_close callback with the about-to-be-closed socket"""
- with self._callback_mutex:
- on_socket_close = self.on_socket_close
-
- if on_socket_close:
- with self._in_callback_mutex:
- try:
- on_socket_close(self, self._userdata, sock)
- except Exception as err:
- self._easy_log(
- MQTT_LOG_ERR, 'Caught exception in on_socket_close: %s', err)
- if not self.suppress_exceptions:
- raise
-
- @property
- def on_socket_register_write(self) -> CallbackOnSocket | None:
- """The callback called when the socket needs writing but can't.
-
- This should be used to register the socket with an external event loop for writing.
-
- Expected signature is (for all callback API version)::
-
- socket_register_write_callback(client, userdata, socket)
-
- :param Client client: the client instance for this callback
- :param userdata: the private user data as set in Client() or user_data_set()
- :param SocketLike sock: the socket which should be registered for writing
-
- Decorator: @client.socket_register_write_callback() (``client`` is the name of the
- instance which this callback is being attached to)
- """
- return self._on_socket_register_write
-
- @on_socket_register_write.setter
- def on_socket_register_write(self, func: CallbackOnSocket | None) -> None:
- with self._callback_mutex:
- self._on_socket_register_write = func
-
- def socket_register_write_callback(
- self,
- ) -> Callable[[CallbackOnSocket], CallbackOnSocket]:
- def decorator(func: CallbackOnSocket) -> CallbackOnSocket:
- self._on_socket_register_write = func
- return func
- return decorator
-
- def _call_socket_register_write(self) -> None:
- """Call the socket_register_write callback with the unwritable socket"""
- if not self._sock or self._registered_write:
- return
- self._registered_write = True
- with self._callback_mutex:
- on_socket_register_write = self.on_socket_register_write
-
- if on_socket_register_write:
- try:
- on_socket_register_write(
- self, self._userdata, self._sock)
- except Exception as err:
- self._easy_log(
- MQTT_LOG_ERR, 'Caught exception in on_socket_register_write: %s', err)
- if not self.suppress_exceptions:
- raise
-
- @property
- def on_socket_unregister_write(
- self,
- ) -> CallbackOnSocket | None:
- """The callback called when the socket doesn't need writing anymore.
-
- This should be used to unregister the socket from an external event loop for writing.
-
- Expected signature is (for all callback API version)::
-
- socket_unregister_write_callback(client, userdata, socket)
-
- :param Client client: the client instance for this callback
- :param userdata: the private user data as set in Client() or user_data_set()
- :param SocketLike sock: the socket which should be unregistered for writing
-
- Decorator: @client.socket_unregister_write_callback() (``client`` is the name of the
- instance which this callback is being attached to)
- """
- return self._on_socket_unregister_write
-
- @on_socket_unregister_write.setter
- def on_socket_unregister_write(
- self, func: CallbackOnSocket | None
- ) -> None:
- with self._callback_mutex:
- self._on_socket_unregister_write = func
-
- def socket_unregister_write_callback(
- self,
- ) -> Callable[[CallbackOnSocket], CallbackOnSocket]:
- def decorator(
- func: CallbackOnSocket,
- ) -> CallbackOnSocket:
- self._on_socket_unregister_write = func
- return func
- return decorator
-
- def _call_socket_unregister_write(
- self, sock: SocketLike | None = None
- ) -> None:
- """Call the socket_unregister_write callback with the writable socket"""
- sock = sock or self._sock
- if not sock or not self._registered_write:
- return
- self._registered_write = False
-
- with self._callback_mutex:
- on_socket_unregister_write = self.on_socket_unregister_write
-
- if on_socket_unregister_write:
- try:
- on_socket_unregister_write(self, self._userdata, sock)
- except Exception as err:
- self._easy_log(
- MQTT_LOG_ERR, 'Caught exception in on_socket_unregister_write: %s', err)
- if not self.suppress_exceptions:
- raise
-
- def message_callback_add(self, sub: str, callback: CallbackOnMessage) -> None:
- """Register a message callback for a specific topic.
- Messages that match 'sub' will be passed to 'callback'. Any
- non-matching messages will be passed to the default `on_message`
- callback.
-
- Call multiple times with different 'sub' to define multiple topic
- specific callbacks.
-
- Topic specific callbacks may be removed with
- `message_callback_remove()`.
-
- See `on_message` for the expected signature of the callback.
-
- Decorator: @client.topic_callback(sub) (``client`` is the name of the
- instance which this callback is being attached to)
-
- Example::
-
- @client.topic_callback("mytopic/#")
- def handle_mytopic(client, userdata, message):
- ...
- """
- if callback is None or sub is None:
- raise ValueError("sub and callback must both be defined.")
-
- with self._callback_mutex:
- self._on_message_filtered[sub] = callback
-
- def topic_callback(
- self, sub: str
- ) -> Callable[[CallbackOnMessage], CallbackOnMessage]:
- def decorator(func: CallbackOnMessage) -> CallbackOnMessage:
- self.message_callback_add(sub, func)
- return func
- return decorator
-
- def message_callback_remove(self, sub: str) -> None:
- """Remove a message callback previously registered with
- `message_callback_add()`."""
- if sub is None:
- raise ValueError("sub must defined.")
-
- with self._callback_mutex:
- try:
- del self._on_message_filtered[sub]
- except KeyError: # no such subscription
- pass
-
- # ============================================================
- # Private functions
- # ============================================================
-
- def _loop_rc_handle(
- self,
- rc: MQTTErrorCode,
- ) -> MQTTErrorCode:
- if rc:
- self._sock_close()
-
- if self._state in (_ConnectionState.MQTT_CS_DISCONNECTING, _ConnectionState.MQTT_CS_DISCONNECTED):
- self._state = _ConnectionState.MQTT_CS_DISCONNECTED
- rc = MQTTErrorCode.MQTT_ERR_SUCCESS
-
- self._do_on_disconnect(packet_from_broker=False, v1_rc=rc)
-
- if rc == MQTT_ERR_CONN_LOST:
- self._state = _ConnectionState.MQTT_CS_CONNECTION_LOST
-
- return rc
-
- def _packet_read(self) -> MQTTErrorCode:
- # This gets called if pselect() indicates that there is network data
- # available - ie. at least one byte. What we do depends on what data we
- # already have.
- # If we've not got a command, attempt to read one and save it. This should
- # always work because it's only a single byte.
- # Then try to read the remaining length. This may fail because it is may
- # be more than one byte - will need to save data pending next read if it
- # does fail.
- # Then try to read the remaining payload, where 'payload' here means the
- # combined variable header and actual payload. This is the most likely to
- # fail due to longer length, so save current data and current position.
- # After all data is read, send to _mqtt_handle_packet() to deal with.
- # Finally, free the memory and reset everything to starting conditions.
- if self._in_packet['command'] == 0:
- try:
- command = self._sock_recv(1)
- except BlockingIOError:
- return MQTTErrorCode.MQTT_ERR_AGAIN
- except TimeoutError as err:
- self._easy_log(
- MQTT_LOG_ERR, 'timeout on socket: %s', err)
- return MQTTErrorCode.MQTT_ERR_CONN_LOST
- except OSError as err:
- self._easy_log(
- MQTT_LOG_ERR, 'failed to receive on socket: %s', err)
- return MQTTErrorCode.MQTT_ERR_CONN_LOST
- else:
- if len(command) == 0:
- return MQTTErrorCode.MQTT_ERR_CONN_LOST
- self._in_packet['command'] = command[0]
-
- if self._in_packet['have_remaining'] == 0:
- # Read remaining
- # Algorithm for decoding taken from pseudo code at
- # http://publib.boulder.ibm.com/infocenter/wmbhelp/v6r0m0/topic/com.ibm.etools.mft.doc/ac10870_.htm
- while True:
- try:
- byte = self._sock_recv(1)
- except BlockingIOError:
- return MQTTErrorCode.MQTT_ERR_AGAIN
- except OSError as err:
- self._easy_log(
- MQTT_LOG_ERR, 'failed to receive on socket: %s', err)
- return MQTTErrorCode.MQTT_ERR_CONN_LOST
- else:
- if len(byte) == 0:
- return MQTTErrorCode.MQTT_ERR_CONN_LOST
- byte_value = byte[0]
- self._in_packet['remaining_count'].append(byte_value)
- # Max 4 bytes length for remaining length as defined by protocol.
- # Anything more likely means a broken/malicious client.
- if len(self._in_packet['remaining_count']) > 4:
- return MQTTErrorCode.MQTT_ERR_PROTOCOL
-
- self._in_packet['remaining_length'] += (
- byte_value & 127) * self._in_packet['remaining_mult']
- self._in_packet['remaining_mult'] = self._in_packet['remaining_mult'] * 128
-
- if (byte_value & 128) == 0:
- break
-
- self._in_packet['have_remaining'] = 1
- self._in_packet['to_process'] = self._in_packet['remaining_length']
-
- count = 100 # Don't get stuck in this loop if we have a huge message.
- while self._in_packet['to_process'] > 0:
- try:
- data = self._sock_recv(self._in_packet['to_process'])
- except BlockingIOError:
- return MQTTErrorCode.MQTT_ERR_AGAIN
- except OSError as err:
- self._easy_log(
- MQTT_LOG_ERR, 'failed to receive on socket: %s', err)
- return MQTTErrorCode.MQTT_ERR_CONN_LOST
- else:
- if len(data) == 0:
- return MQTTErrorCode.MQTT_ERR_CONN_LOST
- self._in_packet['to_process'] -= len(data)
- self._in_packet['packet'] += data
- count -= 1
- if count == 0:
- with self._msgtime_mutex:
- self._last_msg_in = time_func()
- return MQTTErrorCode.MQTT_ERR_AGAIN
-
- # All data for this packet is read.
- self._in_packet['pos'] = 0
- rc = self._packet_handle()
-
- # Free data and reset values
- self._in_packet = {
- "command": 0,
- "have_remaining": 0,
- "remaining_count": [],
- "remaining_mult": 1,
- "remaining_length": 0,
- "packet": bytearray(b""),
- "to_process": 0,
- "pos": 0,
- }
-
- with self._msgtime_mutex:
- self._last_msg_in = time_func()
- return rc
-
- def _packet_write(self) -> MQTTErrorCode:
- while True:
- try:
- packet = self._out_packet.popleft()
- except IndexError:
- return MQTTErrorCode.MQTT_ERR_SUCCESS
-
- try:
- write_length = self._sock_send(
- packet['packet'][packet['pos']:])
- except (AttributeError, ValueError):
- self._out_packet.appendleft(packet)
- return MQTTErrorCode.MQTT_ERR_SUCCESS
- except BlockingIOError:
- self._out_packet.appendleft(packet)
- return MQTTErrorCode.MQTT_ERR_AGAIN
- except OSError as err:
- self._out_packet.appendleft(packet)
- self._easy_log(
- MQTT_LOG_ERR, 'failed to receive on socket: %s', err)
- return MQTTErrorCode.MQTT_ERR_CONN_LOST
-
- if write_length > 0:
- packet['to_process'] -= write_length
- packet['pos'] += write_length
-
- if packet['to_process'] == 0:
- if (packet['command'] & 0xF0) == PUBLISH and packet['qos'] == 0:
- with self._callback_mutex:
- on_publish = self.on_publish
-
- if on_publish:
- with self._in_callback_mutex:
- try:
- if self._callback_api_version == CallbackAPIVersion.VERSION1:
- on_publish = cast(CallbackOnPublish_v1, on_publish)
-
- on_publish(self, self._userdata, packet["mid"])
- elif self._callback_api_version == CallbackAPIVersion.VERSION2:
- on_publish = cast(CallbackOnPublish_v2, on_publish)
-
- on_publish(
- self,
- self._userdata,
- packet["mid"],
- ReasonCode(PacketTypes.PUBACK),
- Properties(PacketTypes.PUBACK),
- )
- else:
- raise RuntimeError("Unsupported callback API version")
- except Exception as err:
- self._easy_log(
- MQTT_LOG_ERR, 'Caught exception in on_publish: %s', err)
- if not self.suppress_exceptions:
- raise
-
- # TODO: Something is odd here. I don't see why packet["info"] can't be None.
- # A packet could be produced by _handle_connack with qos=0 and no info
- # (around line 3645). Ignore the mypy check for now but I feel there is a bug
- # somewhere.
- packet['info']._set_as_published() # type: ignore
-
- if (packet['command'] & 0xF0) == DISCONNECT:
- with self._msgtime_mutex:
- self._last_msg_out = time_func()
-
- self._do_on_disconnect(
- packet_from_broker=False,
- v1_rc=MQTTErrorCode.MQTT_ERR_SUCCESS,
- )
- self._sock_close()
- # Only change to disconnected if the disconnection was wanted
- # by the client (== state was disconnecting). If the broker disconnected
- # use unilaterally don't change the state and client may reconnect.
- if self._state == _ConnectionState.MQTT_CS_DISCONNECTING:
- self._state = _ConnectionState.MQTT_CS_DISCONNECTED
- return MQTTErrorCode.MQTT_ERR_SUCCESS
-
- else:
- # We haven't finished with this packet
- self._out_packet.appendleft(packet)
- else:
- break
-
- with self._msgtime_mutex:
- self._last_msg_out = time_func()
-
- return MQTTErrorCode.MQTT_ERR_SUCCESS
-
- def _easy_log(self, level: LogLevel, fmt: str, *args: Any) -> None:
- if self.on_log is not None:
- buf = fmt % args
- try:
- self.on_log(self, self._userdata, level, buf)
- except Exception: # noqa: S110
- # Can't _easy_log this, as we'll recurse until we break
- pass # self._logger will pick this up, so we're fine
- if self._logger is not None:
- level_std = LOGGING_LEVEL[level]
- self._logger.log(level_std, fmt, *args)
-
- def _check_keepalive(self) -> None:
- if self._keepalive == 0:
- return
-
- now = time_func()
-
- with self._msgtime_mutex:
- last_msg_out = self._last_msg_out
- last_msg_in = self._last_msg_in
-
- if self._sock is not None and (now - last_msg_out >= self._keepalive or now - last_msg_in >= self._keepalive):
- if self._state == _ConnectionState.MQTT_CS_CONNECTED and self._ping_t == 0:
- try:
- self._send_pingreq()
- except Exception:
- self._sock_close()
- self._do_on_disconnect(
- packet_from_broker=False,
- v1_rc=MQTTErrorCode.MQTT_ERR_CONN_LOST,
- )
- else:
- with self._msgtime_mutex:
- self._last_msg_out = now
- self._last_msg_in = now
- else:
- self._sock_close()
-
- if self._state in (_ConnectionState.MQTT_CS_DISCONNECTING, _ConnectionState.MQTT_CS_DISCONNECTED):
- self._state = _ConnectionState.MQTT_CS_DISCONNECTED
- rc = MQTTErrorCode.MQTT_ERR_SUCCESS
- else:
- rc = MQTTErrorCode.MQTT_ERR_KEEPALIVE
-
- self._do_on_disconnect(
- packet_from_broker=False,
- v1_rc=rc,
- )
-
- def _mid_generate(self) -> int:
- with self._mid_generate_mutex:
- self._last_mid += 1
- if self._last_mid == 65536:
- self._last_mid = 1
- return self._last_mid
-
- @staticmethod
- def _raise_for_invalid_topic(topic: bytes) -> None:
- """ Check if the topic is a topic without wildcard and valid length.
-
- Raise ValueError if the topic isn't valid.
- """
- if b'+' in topic or b'#' in topic:
- raise ValueError('Publish topic cannot contain wildcards.')
- if len(topic) > 65535:
- raise ValueError('Publish topic is too long.')
-
- @staticmethod
- def _filter_wildcard_len_check(sub: bytes) -> MQTTErrorCode:
- if (len(sub) == 0 or len(sub) > 65535
- or any(b'+' in p or b'#' in p for p in sub.split(b'/') if len(p) > 1)
- or b'#/' in sub):
- return MQTTErrorCode.MQTT_ERR_INVAL
- else:
- return MQTTErrorCode.MQTT_ERR_SUCCESS
-
- def _send_pingreq(self) -> MQTTErrorCode:
- self._easy_log(MQTT_LOG_DEBUG, "Sending PINGREQ")
- rc = self._send_simple_command(PINGREQ)
- if rc == MQTTErrorCode.MQTT_ERR_SUCCESS:
- self._ping_t = time_func()
- return rc
-
- def _send_pingresp(self) -> MQTTErrorCode:
- self._easy_log(MQTT_LOG_DEBUG, "Sending PINGRESP")
- return self._send_simple_command(PINGRESP)
-
- def _send_puback(self, mid: int) -> MQTTErrorCode:
- self._easy_log(MQTT_LOG_DEBUG, "Sending PUBACK (Mid: %d)", mid)
- return self._send_command_with_mid(PUBACK, mid, False)
-
- def _send_pubcomp(self, mid: int) -> MQTTErrorCode:
- self._easy_log(MQTT_LOG_DEBUG, "Sending PUBCOMP (Mid: %d)", mid)
- return self._send_command_with_mid(PUBCOMP, mid, False)
-
- def _pack_remaining_length(
- self, packet: bytearray, remaining_length: int
- ) -> bytearray:
- remaining_bytes = []
- while True:
- byte = remaining_length % 128
- remaining_length = remaining_length // 128
- # If there are more digits to encode, set the top bit of this digit
- if remaining_length > 0:
- byte |= 0x80
-
- remaining_bytes.append(byte)
- packet.append(byte)
- if remaining_length == 0:
- # FIXME - this doesn't deal with incorrectly large payloads
- return packet
-
- def _pack_str16(self, packet: bytearray, data: bytes | str) -> None:
- data = _force_bytes(data)
- packet.extend(struct.pack("!H", len(data)))
- packet.extend(data)
-
- def _send_publish(
- self,
- mid: int,
- topic: bytes,
- payload: bytes|bytearray = b"",
- qos: int = 0,
- retain: bool = False,
- dup: bool = False,
- info: MQTTMessageInfo | None = None,
- properties: Properties | None = None,
- ) -> MQTTErrorCode:
- # we assume that topic and payload are already properly encoded
- if not isinstance(topic, bytes):
- raise TypeError('topic must be bytes, not str')
- if payload and not isinstance(payload, (bytes, bytearray)):
- raise TypeError('payload must be bytes if set')
-
- if self._sock is None:
- return MQTTErrorCode.MQTT_ERR_NO_CONN
-
- command = PUBLISH | ((dup & 0x1) << 3) | (qos << 1) | retain
- packet = bytearray()
- packet.append(command)
-
- payloadlen = len(payload)
- remaining_length = 2 + len(topic) + payloadlen
-
- if payloadlen == 0:
- if self._protocol == MQTTv5:
- self._easy_log(
- MQTT_LOG_DEBUG,
- "Sending PUBLISH (d%d, q%d, r%d, m%d), '%s', properties=%s (NULL payload)",
- dup, qos, retain, mid, topic, properties
- )
- else:
- self._easy_log(
- MQTT_LOG_DEBUG,
- "Sending PUBLISH (d%d, q%d, r%d, m%d), '%s' (NULL payload)",
- dup, qos, retain, mid, topic
- )
- else:
- if self._protocol == MQTTv5:
- self._easy_log(
- MQTT_LOG_DEBUG,
- "Sending PUBLISH (d%d, q%d, r%d, m%d), '%s', properties=%s, ... (%d bytes)",
- dup, qos, retain, mid, topic, properties, payloadlen
- )
- else:
- self._easy_log(
- MQTT_LOG_DEBUG,
- "Sending PUBLISH (d%d, q%d, r%d, m%d), '%s', ... (%d bytes)",
- dup, qos, retain, mid, topic, payloadlen
- )
-
- if qos > 0:
- # For message id
- remaining_length += 2
-
- if self._protocol == MQTTv5:
- if properties is None:
- packed_properties = b'\x00'
- else:
- packed_properties = properties.pack()
- remaining_length += len(packed_properties)
-
- self._pack_remaining_length(packet, remaining_length)
- self._pack_str16(packet, topic)
-
- if qos > 0:
- # For message id
- packet.extend(struct.pack("!H", mid))
-
- if self._protocol == MQTTv5:
- packet.extend(packed_properties)
-
- packet.extend(payload)
-
- return self._packet_queue(PUBLISH, packet, mid, qos, info)
-
- def _send_pubrec(self, mid: int) -> MQTTErrorCode:
- self._easy_log(MQTT_LOG_DEBUG, "Sending PUBREC (Mid: %d)", mid)
- return self._send_command_with_mid(PUBREC, mid, False)
-
- def _send_pubrel(self, mid: int) -> MQTTErrorCode:
- self._easy_log(MQTT_LOG_DEBUG, "Sending PUBREL (Mid: %d)", mid)
- return self._send_command_with_mid(PUBREL | 2, mid, False)
-
- def _send_command_with_mid(self, command: int, mid: int, dup: int) -> MQTTErrorCode:
- # For PUBACK, PUBCOMP, PUBREC, and PUBREL
- if dup:
- command |= 0x8
-
- remaining_length = 2
- packet = struct.pack('!BBH', command, remaining_length, mid)
- return self._packet_queue(command, packet, mid, 1)
-
- def _send_simple_command(self, command: int) -> MQTTErrorCode:
- # For DISCONNECT, PINGREQ and PINGRESP
- remaining_length = 0
- packet = struct.pack('!BB', command, remaining_length)
- return self._packet_queue(command, packet, 0, 0)
-
- def _send_connect(self, keepalive: int) -> MQTTErrorCode:
- proto_ver = int(self._protocol)
- # hard-coded UTF-8 encoded string
- protocol = b"MQTT" if proto_ver >= MQTTv311 else b"MQIsdp"
-
- remaining_length = 2 + len(protocol) + 1 + \
- 1 + 2 + 2 + len(self._client_id)
-
- connect_flags = 0
- if self._protocol == MQTTv5:
- if self._clean_start is True:
- connect_flags |= 0x02
- elif self._clean_start == MQTT_CLEAN_START_FIRST_ONLY and self._mqttv5_first_connect:
- connect_flags |= 0x02
- elif self._clean_session:
- connect_flags |= 0x02
-
- if self._will:
- remaining_length += 2 + \
- len(self._will_topic) + 2 + len(self._will_payload)
- connect_flags |= 0x04 | ((self._will_qos & 0x03) << 3) | (
- (self._will_retain & 0x01) << 5)
-
- if self._username is not None:
- remaining_length += 2 + len(self._username)
- connect_flags |= 0x80
- if self._password is not None:
- connect_flags |= 0x40
- remaining_length += 2 + len(self._password)
-
- if self._protocol == MQTTv5:
- if self._connect_properties is None:
- packed_connect_properties = b'\x00'
- else:
- packed_connect_properties = self._connect_properties.pack()
- remaining_length += len(packed_connect_properties)
- if self._will:
- if self._will_properties is None:
- packed_will_properties = b'\x00'
- else:
- packed_will_properties = self._will_properties.pack()
- remaining_length += len(packed_will_properties)
-
- command = CONNECT
- packet = bytearray()
- packet.append(command)
-
- # as per the mosquitto broker, if the MSB of this version is set
- # to 1, then it treats the connection as a bridge
- if self._client_mode == MQTT_BRIDGE:
- proto_ver |= 0x80
-
- self._pack_remaining_length(packet, remaining_length)
- packet.extend(struct.pack(
- f"!H{len(protocol)}sBBH",
- len(protocol), protocol, proto_ver, connect_flags, keepalive,
- ))
-
- if self._protocol == MQTTv5:
- packet += packed_connect_properties
-
- self._pack_str16(packet, self._client_id)
-
- if self._will:
- if self._protocol == MQTTv5:
- packet += packed_will_properties
- self._pack_str16(packet, self._will_topic)
- self._pack_str16(packet, self._will_payload)
-
- if self._username is not None:
- self._pack_str16(packet, self._username)
-
- if self._password is not None:
- self._pack_str16(packet, self._password)
-
- self._keepalive = keepalive
- if self._protocol == MQTTv5:
- self._easy_log(
- MQTT_LOG_DEBUG,
- "Sending CONNECT (u%d, p%d, wr%d, wq%d, wf%d, c%d, k%d) client_id=%s properties=%s",
- (connect_flags & 0x80) >> 7,
- (connect_flags & 0x40) >> 6,
- (connect_flags & 0x20) >> 5,
- (connect_flags & 0x18) >> 3,
- (connect_flags & 0x4) >> 2,
- (connect_flags & 0x2) >> 1,
- keepalive,
- self._client_id,
- self._connect_properties
- )
- else:
- self._easy_log(
- MQTT_LOG_DEBUG,
- "Sending CONNECT (u%d, p%d, wr%d, wq%d, wf%d, c%d, k%d) client_id=%s",
- (connect_flags & 0x80) >> 7,
- (connect_flags & 0x40) >> 6,
- (connect_flags & 0x20) >> 5,
- (connect_flags & 0x18) >> 3,
- (connect_flags & 0x4) >> 2,
- (connect_flags & 0x2) >> 1,
- keepalive,
- self._client_id
- )
- return self._packet_queue(command, packet, 0, 0)
-
- def _send_disconnect(
- self,
- reasoncode: ReasonCode | None = None,
- properties: Properties | None = None,
- ) -> MQTTErrorCode:
- if self._protocol == MQTTv5:
- self._easy_log(MQTT_LOG_DEBUG, "Sending DISCONNECT reasonCode=%s properties=%s",
- reasoncode,
- properties
- )
- else:
- self._easy_log(MQTT_LOG_DEBUG, "Sending DISCONNECT")
-
- remaining_length = 0
-
- command = DISCONNECT
- packet = bytearray()
- packet.append(command)
-
- if self._protocol == MQTTv5:
- if properties is not None or reasoncode is not None:
- if reasoncode is None:
- reasoncode = ReasonCode(DISCONNECT >> 4, identifier=0)
- remaining_length += 1
- if properties is not None:
- packed_props = properties.pack()
- remaining_length += len(packed_props)
-
- self._pack_remaining_length(packet, remaining_length)
-
- if self._protocol == MQTTv5:
- if reasoncode is not None:
- packet += reasoncode.pack()
- if properties is not None:
- packet += packed_props
-
- return self._packet_queue(command, packet, 0, 0)
-
- def _send_subscribe(
- self,
- dup: int,
- topics: Sequence[tuple[bytes, SubscribeOptions | int]],
- properties: Properties | None = None,
- ) -> tuple[MQTTErrorCode, int]:
- remaining_length = 2
- if self._protocol == MQTTv5:
- if properties is None:
- packed_subscribe_properties = b'\x00'
- else:
- packed_subscribe_properties = properties.pack()
- remaining_length += len(packed_subscribe_properties)
- for t, _ in topics:
- remaining_length += 2 + len(t) + 1
-
- command = SUBSCRIBE | (dup << 3) | 0x2
- packet = bytearray()
- packet.append(command)
- self._pack_remaining_length(packet, remaining_length)
- local_mid = self._mid_generate()
- packet.extend(struct.pack("!H", local_mid))
-
- if self._protocol == MQTTv5:
- packet += packed_subscribe_properties
-
- for t, q in topics:
- self._pack_str16(packet, t)
- if self._protocol == MQTTv5:
- packet += q.pack() # type: ignore
- else:
- packet.append(q) # type: ignore
-
- self._easy_log(
- MQTT_LOG_DEBUG,
- "Sending SUBSCRIBE (d%d, m%d) %s",
- dup,
- local_mid,
- topics,
- )
- return (self._packet_queue(command, packet, local_mid, 1), local_mid)
-
- def _send_unsubscribe(
- self,
- dup: int,
- topics: list[bytes],
- properties: Properties | None = None,
- ) -> tuple[MQTTErrorCode, int]:
- remaining_length = 2
- if self._protocol == MQTTv5:
- if properties is None:
- packed_unsubscribe_properties = b'\x00'
- else:
- packed_unsubscribe_properties = properties.pack()
- remaining_length += len(packed_unsubscribe_properties)
- for t in topics:
- remaining_length += 2 + len(t)
-
- command = UNSUBSCRIBE | (dup << 3) | 0x2
- packet = bytearray()
- packet.append(command)
- self._pack_remaining_length(packet, remaining_length)
- local_mid = self._mid_generate()
- packet.extend(struct.pack("!H", local_mid))
-
- if self._protocol == MQTTv5:
- packet += packed_unsubscribe_properties
-
- for t in topics:
- self._pack_str16(packet, t)
-
- # topics_repr = ", ".join("'"+topic.decode('utf8')+"'" for topic in topics)
- if self._protocol == MQTTv5:
- self._easy_log(
- MQTT_LOG_DEBUG,
- "Sending UNSUBSCRIBE (d%d, m%d) %s %s",
- dup,
- local_mid,
- properties,
- topics,
- )
- else:
- self._easy_log(
- MQTT_LOG_DEBUG,
- "Sending UNSUBSCRIBE (d%d, m%d) %s",
- dup,
- local_mid,
- topics,
- )
- return (self._packet_queue(command, packet, local_mid, 1), local_mid)
-
- def _check_clean_session(self) -> bool:
- if self._protocol == MQTTv5:
- if self._clean_start == MQTT_CLEAN_START_FIRST_ONLY:
- return self._mqttv5_first_connect
- else:
- return self._clean_start # type: ignore
- else:
- return self._clean_session
-
- def _messages_reconnect_reset_out(self) -> None:
- with self._out_message_mutex:
- self._inflight_messages = 0
- for m in self._out_messages.values():
- m.timestamp = 0
- if self._max_inflight_messages == 0 or self._inflight_messages < self._max_inflight_messages:
- if m.qos == 0:
- m.state = mqtt_ms_publish
- elif m.qos == 1:
- # self._inflight_messages = self._inflight_messages + 1
- if m.state == mqtt_ms_wait_for_puback:
- m.dup = True
- m.state = mqtt_ms_publish
- elif m.qos == 2:
- # self._inflight_messages = self._inflight_messages + 1
- if self._check_clean_session():
- if m.state != mqtt_ms_publish:
- m.dup = True
- m.state = mqtt_ms_publish
- else:
- if m.state == mqtt_ms_wait_for_pubcomp:
- m.state = mqtt_ms_resend_pubrel
- else:
- if m.state == mqtt_ms_wait_for_pubrec:
- m.dup = True
- m.state = mqtt_ms_publish
- else:
- m.state = mqtt_ms_queued
-
- def _messages_reconnect_reset_in(self) -> None:
- with self._in_message_mutex:
- if self._check_clean_session():
- self._in_messages = collections.OrderedDict()
- return
- for m in self._in_messages.values():
- m.timestamp = 0
- if m.qos != 2:
- self._in_messages.pop(m.mid)
- else:
- # Preserve current state
- pass
-
- def _messages_reconnect_reset(self) -> None:
- self._messages_reconnect_reset_out()
- self._messages_reconnect_reset_in()
-
- def _packet_queue(
- self,
- command: int,
- packet: bytes,
- mid: int,
- qos: int,
- info: MQTTMessageInfo | None = None,
- ) -> MQTTErrorCode:
- mpkt: _OutPacket = {
- "command": command,
- "mid": mid,
- "qos": qos,
- "pos": 0,
- "to_process": len(packet),
- "packet": packet,
- "info": info,
- }
-
- self._out_packet.append(mpkt)
-
- # Write a single byte to sockpairW (connected to sockpairR) to break
- # out of select() if in threaded mode.
- if self._sockpairW is not None:
- try:
- self._sockpairW.send(sockpair_data)
- except BlockingIOError:
- pass
-
- # If we have an external event loop registered, use that instead
- # of calling loop_write() directly.
- if self._thread is None and self._on_socket_register_write is None:
- if self._in_callback_mutex.acquire(False):
- self._in_callback_mutex.release()
- return self.loop_write()
-
- self._call_socket_register_write()
-
- return MQTTErrorCode.MQTT_ERR_SUCCESS
-
- def _packet_handle(self) -> MQTTErrorCode:
- cmd = self._in_packet['command'] & 0xF0
- if cmd == PINGREQ:
- return self._handle_pingreq()
- elif cmd == PINGRESP:
- return self._handle_pingresp()
- elif cmd == PUBACK:
- return self._handle_pubackcomp("PUBACK")
- elif cmd == PUBCOMP:
- return self._handle_pubackcomp("PUBCOMP")
- elif cmd == PUBLISH:
- return self._handle_publish()
- elif cmd == PUBREC:
- return self._handle_pubrec()
- elif cmd == PUBREL:
- return self._handle_pubrel()
- elif cmd == CONNACK:
- return self._handle_connack()
- elif cmd == SUBACK:
- self._handle_suback()
- return MQTTErrorCode.MQTT_ERR_SUCCESS
- elif cmd == UNSUBACK:
- return self._handle_unsuback()
- elif cmd == DISCONNECT and self._protocol == MQTTv5: # only allowed in MQTT 5.0
- self._handle_disconnect()
- return MQTTErrorCode.MQTT_ERR_SUCCESS
- else:
- # If we don't recognise the command, return an error straight away.
- self._easy_log(MQTT_LOG_ERR, "Error: Unrecognised command %s", cmd)
- return MQTTErrorCode.MQTT_ERR_PROTOCOL
-
- def _handle_pingreq(self) -> MQTTErrorCode:
- if self._in_packet['remaining_length'] != 0:
- return MQTTErrorCode.MQTT_ERR_PROTOCOL
-
- self._easy_log(MQTT_LOG_DEBUG, "Received PINGREQ")
- return self._send_pingresp()
-
- def _handle_pingresp(self) -> MQTTErrorCode:
- if self._in_packet['remaining_length'] != 0:
- return MQTTErrorCode.MQTT_ERR_PROTOCOL
-
- # No longer waiting for a PINGRESP.
- self._ping_t = 0
- self._easy_log(MQTT_LOG_DEBUG, "Received PINGRESP")
- return MQTTErrorCode.MQTT_ERR_SUCCESS
-
- def _handle_connack(self) -> MQTTErrorCode:
- if self._protocol == MQTTv5:
- if self._in_packet['remaining_length'] < 2:
- return MQTTErrorCode.MQTT_ERR_PROTOCOL
- elif self._in_packet['remaining_length'] != 2:
- return MQTTErrorCode.MQTT_ERR_PROTOCOL
-
- if self._protocol == MQTTv5:
- (flags, result) = struct.unpack(
- "!BB", self._in_packet['packet'][:2])
- if result == 1:
- # This is probably a failure from a broker that doesn't support
- # MQTT v5.
- reason = ReasonCode(CONNACK >> 4, aName="Unsupported protocol version")
- properties = None
- else:
- reason = ReasonCode(CONNACK >> 4, identifier=result)
- properties = Properties(CONNACK >> 4)
- properties.unpack(self._in_packet['packet'][2:])
- else:
- (flags, result) = struct.unpack("!BB", self._in_packet['packet'])
- reason = convert_connack_rc_to_reason_code(result)
- properties = None
- if self._protocol == MQTTv311:
- if result == CONNACK_REFUSED_PROTOCOL_VERSION:
- if not self._reconnect_on_failure:
- return MQTT_ERR_PROTOCOL
- self._easy_log(
- MQTT_LOG_DEBUG,
- "Received CONNACK (%s, %s), attempting downgrade to MQTT v3.1.",
- flags, result
- )
- # Downgrade to MQTT v3.1
- self._protocol = MQTTv31
- return self.reconnect()
- elif (result == CONNACK_REFUSED_IDENTIFIER_REJECTED
- and self._client_id == b''):
- if not self._reconnect_on_failure:
- return MQTT_ERR_PROTOCOL
- self._easy_log(
- MQTT_LOG_DEBUG,
- "Received CONNACK (%s, %s), attempting to use non-empty CID",
- flags, result,
- )
- self._client_id = _base62(uuid.uuid4().int, padding=22).encode("utf8")
- return self.reconnect()
-
- if result == 0:
- self._state = _ConnectionState.MQTT_CS_CONNECTED
- self._reconnect_delay = None
-
- if self._protocol == MQTTv5:
- self._easy_log(
- MQTT_LOG_DEBUG, "Received CONNACK (%s, %s) properties=%s", flags, reason, properties)
- else:
- self._easy_log(
- MQTT_LOG_DEBUG, "Received CONNACK (%s, %s)", flags, result)
-
- # it won't be the first successful connect any more
- self._mqttv5_first_connect = False
-
- with self._callback_mutex:
- on_connect = self.on_connect
-
- if on_connect:
- flags_dict = {}
- flags_dict['session present'] = flags & 0x01
- with self._in_callback_mutex:
- try:
- if self._callback_api_version == CallbackAPIVersion.VERSION1:
- if self._protocol == MQTTv5:
- on_connect = cast(CallbackOnConnect_v1_mqtt5, on_connect)
-
- on_connect(self, self._userdata,
- flags_dict, reason, properties)
- else:
- on_connect = cast(CallbackOnConnect_v1_mqtt3, on_connect)
-
- on_connect(
- self, self._userdata, flags_dict, result)
- elif self._callback_api_version == CallbackAPIVersion.VERSION2:
- on_connect = cast(CallbackOnConnect_v2, on_connect)
-
- connect_flags = ConnectFlags(
- session_present=flags_dict['session present'] > 0
- )
-
- if properties is None:
- properties = Properties(PacketTypes.CONNACK)
-
- on_connect(
- self,
- self._userdata,
- connect_flags,
- reason,
- properties,
- )
- else:
- raise RuntimeError("Unsupported callback API version")
- except Exception as err:
- self._easy_log(
- MQTT_LOG_ERR, 'Caught exception in on_connect: %s', err)
- if not self.suppress_exceptions:
- raise
-
- if result == 0:
- rc = MQTTErrorCode.MQTT_ERR_SUCCESS
- with self._out_message_mutex:
- for m in self._out_messages.values():
- m.timestamp = time_func()
- if m.state == mqtt_ms_queued:
- self.loop_write() # Process outgoing messages that have just been queued up
- return MQTT_ERR_SUCCESS
-
- if m.qos == 0:
- with self._in_callback_mutex: # Don't call loop_write after _send_publish()
- rc = self._send_publish(
- m.mid,
- m.topic.encode('utf-8'),
- m.payload,
- m.qos,
- m.retain,
- m.dup,
- properties=m.properties
- )
- if rc != MQTTErrorCode.MQTT_ERR_SUCCESS:
- return rc
- elif m.qos == 1:
- if m.state == mqtt_ms_publish:
- self._inflight_messages += 1
- m.state = mqtt_ms_wait_for_puback
- with self._in_callback_mutex: # Don't call loop_write after _send_publish()
- rc = self._send_publish(
- m.mid,
- m.topic.encode('utf-8'),
- m.payload,
- m.qos,
- m.retain,
- m.dup,
- properties=m.properties
- )
- if rc != MQTTErrorCode.MQTT_ERR_SUCCESS:
- return rc
- elif m.qos == 2:
- if m.state == mqtt_ms_publish:
- self._inflight_messages += 1
- m.state = mqtt_ms_wait_for_pubrec
- with self._in_callback_mutex: # Don't call loop_write after _send_publish()
- rc = self._send_publish(
- m.mid,
- m.topic.encode('utf-8'),
- m.payload,
- m.qos,
- m.retain,
- m.dup,
- properties=m.properties
- )
- if rc != MQTTErrorCode.MQTT_ERR_SUCCESS:
- return rc
- elif m.state == mqtt_ms_resend_pubrel:
- self._inflight_messages += 1
- m.state = mqtt_ms_wait_for_pubcomp
- with self._in_callback_mutex: # Don't call loop_write after _send_publish()
- rc = self._send_pubrel(m.mid)
- if rc != MQTTErrorCode.MQTT_ERR_SUCCESS:
- return rc
- self.loop_write() # Process outgoing messages that have just been queued up
-
- return rc
- elif result > 0 and result < 6:
- return MQTTErrorCode.MQTT_ERR_CONN_REFUSED
- else:
- return MQTTErrorCode.MQTT_ERR_PROTOCOL
-
- def _handle_disconnect(self) -> None:
- packet_type = DISCONNECT >> 4
- reasonCode = properties = None
- if self._in_packet['remaining_length'] > 2:
- reasonCode = ReasonCode(packet_type)
- reasonCode.unpack(self._in_packet['packet'])
- if self._in_packet['remaining_length'] > 3:
- properties = Properties(packet_type)
- props, props_len = properties.unpack(
- self._in_packet['packet'][1:])
- self._easy_log(MQTT_LOG_DEBUG, "Received DISCONNECT %s %s",
- reasonCode,
- properties
- )
-
- self._sock_close()
- self._do_on_disconnect(
- packet_from_broker=True,
- v1_rc=MQTTErrorCode.MQTT_ERR_SUCCESS, # If reason is absent (remaining length < 1), it means normal disconnection
- reason=reasonCode,
- properties=properties,
- )
-
- def _handle_suback(self) -> None:
- self._easy_log(MQTT_LOG_DEBUG, "Received SUBACK")
- pack_format = f"!H{len(self._in_packet['packet']) - 2}s"
- (mid, packet) = struct.unpack(pack_format, self._in_packet['packet'])
-
- if self._protocol == MQTTv5:
- properties = Properties(SUBACK >> 4)
- props, props_len = properties.unpack(packet)
- reasoncodes = [ReasonCode(SUBACK >> 4, identifier=c) for c in packet[props_len:]]
- else:
- pack_format = f"!{'B' * len(packet)}"
- granted_qos = struct.unpack(pack_format, packet)
- reasoncodes = [ReasonCode(SUBACK >> 4, identifier=c) for c in granted_qos]
- properties = Properties(SUBACK >> 4)
-
- with self._callback_mutex:
- on_subscribe = self.on_subscribe
-
- if on_subscribe:
- with self._in_callback_mutex: # Don't call loop_write after _send_publish()
- try:
- if self._callback_api_version == CallbackAPIVersion.VERSION1:
- if self._protocol == MQTTv5:
- on_subscribe = cast(CallbackOnSubscribe_v1_mqtt5, on_subscribe)
-
- on_subscribe(
- self, self._userdata, mid, reasoncodes, properties)
- else:
- on_subscribe = cast(CallbackOnSubscribe_v1_mqtt3, on_subscribe)
-
- on_subscribe(
- self, self._userdata, mid, granted_qos)
- elif self._callback_api_version == CallbackAPIVersion.VERSION2:
- on_subscribe = cast(CallbackOnSubscribe_v2, on_subscribe)
-
- on_subscribe(
- self,
- self._userdata,
- mid,
- reasoncodes,
- properties,
- )
- else:
- raise RuntimeError("Unsupported callback API version")
- except Exception as err:
- self._easy_log(
- MQTT_LOG_ERR, 'Caught exception in on_subscribe: %s', err)
- if not self.suppress_exceptions:
- raise
-
- def _handle_publish(self) -> MQTTErrorCode:
- header = self._in_packet['command']
- message = MQTTMessage()
- message.dup = ((header & 0x08) >> 3) != 0
- message.qos = (header & 0x06) >> 1
- message.retain = (header & 0x01) != 0
-
- pack_format = f"!H{len(self._in_packet['packet']) - 2}s"
- (slen, packet) = struct.unpack(pack_format, self._in_packet['packet'])
- pack_format = f"!{slen}s{len(packet) - slen}s"
- (topic, packet) = struct.unpack(pack_format, packet)
-
- if self._protocol != MQTTv5 and len(topic) == 0:
- return MQTTErrorCode.MQTT_ERR_PROTOCOL
-
- # Handle topics with invalid UTF-8
- # This replaces an invalid topic with a message and the hex
- # representation of the topic for logging. When the user attempts to
- # access message.topic in the callback, an exception will be raised.
- try:
- print_topic = topic.decode('utf-8')
- except UnicodeDecodeError:
- print_topic = f"TOPIC WITH INVALID UTF-8: {topic!r}"
-
- message.topic = topic
-
- if message.qos > 0:
- pack_format = f"!H{len(packet) - 2}s"
- (message.mid, packet) = struct.unpack(pack_format, packet)
-
- if self._protocol == MQTTv5:
- message.properties = Properties(PUBLISH >> 4)
- props, props_len = message.properties.unpack(packet)
- packet = packet[props_len:]
-
- message.payload = packet
-
- if self._protocol == MQTTv5:
- self._easy_log(
- MQTT_LOG_DEBUG,
- "Received PUBLISH (d%d, q%d, r%d, m%d), '%s', properties=%s, ... (%d bytes)",
- message.dup, message.qos, message.retain, message.mid,
- print_topic, message.properties, len(message.payload)
- )
- else:
- self._easy_log(
- MQTT_LOG_DEBUG,
- "Received PUBLISH (d%d, q%d, r%d, m%d), '%s', ... (%d bytes)",
- message.dup, message.qos, message.retain, message.mid,
- print_topic, len(message.payload)
- )
-
- message.timestamp = time_func()
- if message.qos == 0:
- self._handle_on_message(message)
- return MQTTErrorCode.MQTT_ERR_SUCCESS
- elif message.qos == 1:
- self._handle_on_message(message)
- if self._manual_ack:
- return MQTTErrorCode.MQTT_ERR_SUCCESS
- else:
- return self._send_puback(message.mid)
- elif message.qos == 2:
-
- rc = self._send_pubrec(message.mid)
-
- message.state = mqtt_ms_wait_for_pubrel
- with self._in_message_mutex:
- self._in_messages[message.mid] = message
-
- return rc
- else:
- return MQTTErrorCode.MQTT_ERR_PROTOCOL
-
- def ack(self, mid: int, qos: int) -> MQTTErrorCode:
- """
- send an acknowledgement for a given message id (stored in :py:attr:`message.mid <MQTTMessage.mid>`).
- only useful in QoS>=1 and ``manual_ack=True`` (option of `Client`)
- """
- if self._manual_ack :
- if qos == 1:
- return self._send_puback(mid)
- elif qos == 2:
- return self._send_pubcomp(mid)
-
- return MQTTErrorCode.MQTT_ERR_SUCCESS
-
- def manual_ack_set(self, on: bool) -> None:
- """
- The paho library normally acknowledges messages as soon as they are delivered to the caller.
- If manual_ack is turned on, then the caller MUST manually acknowledge every message once
- application processing is complete using `ack()`
- """
- self._manual_ack = on
-
-
- def _handle_pubrel(self) -> MQTTErrorCode:
- if self._protocol == MQTTv5:
- if self._in_packet['remaining_length'] < 2:
- return MQTTErrorCode.MQTT_ERR_PROTOCOL
- elif self._in_packet['remaining_length'] != 2:
- return MQTTErrorCode.MQTT_ERR_PROTOCOL
-
- mid, = struct.unpack("!H", self._in_packet['packet'][:2])
- if self._protocol == MQTTv5:
- if self._in_packet['remaining_length'] > 2:
- reasonCode = ReasonCode(PUBREL >> 4)
- reasonCode.unpack(self._in_packet['packet'][2:])
- if self._in_packet['remaining_length'] > 3:
- properties = Properties(PUBREL >> 4)
- props, props_len = properties.unpack(
- self._in_packet['packet'][3:])
- self._easy_log(MQTT_LOG_DEBUG, "Received PUBREL (Mid: %d)", mid)
-
- with self._in_message_mutex:
- if mid in self._in_messages:
- # Only pass the message on if we have removed it from the queue - this
- # prevents multiple callbacks for the same message.
- message = self._in_messages.pop(mid)
- self._handle_on_message(message)
- self._inflight_messages -= 1
- if self._max_inflight_messages > 0:
- with self._out_message_mutex:
- rc = self._update_inflight()
- if rc != MQTTErrorCode.MQTT_ERR_SUCCESS:
- return rc
-
- # FIXME: this should only be done if the message is known
- # If unknown it's a protocol error and we should close the connection.
- # But since we don't have (on disk) persistence for the session, it
- # is possible that we must known about this message.
- # Choose to acknowledge this message (thus losing a message) but
- # avoid hanging. See #284.
- if self._manual_ack:
- return MQTTErrorCode.MQTT_ERR_SUCCESS
- else:
- return self._send_pubcomp(mid)
-
- def _update_inflight(self) -> MQTTErrorCode:
- # Dont lock message_mutex here
- for m in self._out_messages.values():
- if self._inflight_messages < self._max_inflight_messages:
- if m.qos > 0 and m.state == mqtt_ms_queued:
- self._inflight_messages += 1
- if m.qos == 1:
- m.state = mqtt_ms_wait_for_puback
- elif m.qos == 2:
- m.state = mqtt_ms_wait_for_pubrec
- rc = self._send_publish(
- m.mid,
- m.topic.encode('utf-8'),
- m.payload,
- m.qos,
- m.retain,
- m.dup,
- properties=m.properties,
- )
- if rc != MQTTErrorCode.MQTT_ERR_SUCCESS:
- return rc
- else:
- return MQTTErrorCode.MQTT_ERR_SUCCESS
- return MQTTErrorCode.MQTT_ERR_SUCCESS
-
- def _handle_pubrec(self) -> MQTTErrorCode:
- if self._protocol == MQTTv5:
- if self._in_packet['remaining_length'] < 2:
- return MQTTErrorCode.MQTT_ERR_PROTOCOL
- elif self._in_packet['remaining_length'] != 2:
- return MQTTErrorCode.MQTT_ERR_PROTOCOL
-
- mid, = struct.unpack("!H", self._in_packet['packet'][:2])
- if self._protocol == MQTTv5:
- if self._in_packet['remaining_length'] > 2:
- reasonCode = ReasonCode(PUBREC >> 4)
- reasonCode.unpack(self._in_packet['packet'][2:])
- if self._in_packet['remaining_length'] > 3:
- properties = Properties(PUBREC >> 4)
- props, props_len = properties.unpack(
- self._in_packet['packet'][3:])
- self._easy_log(MQTT_LOG_DEBUG, "Received PUBREC (Mid: %d)", mid)
-
- with self._out_message_mutex:
- if mid in self._out_messages:
- msg = self._out_messages[mid]
- msg.state = mqtt_ms_wait_for_pubcomp
- msg.timestamp = time_func()
- return self._send_pubrel(mid)
-
- return MQTTErrorCode.MQTT_ERR_SUCCESS
-
- def _handle_unsuback(self) -> MQTTErrorCode:
- if self._protocol == MQTTv5:
- if self._in_packet['remaining_length'] < 4:
- return MQTTErrorCode.MQTT_ERR_PROTOCOL
- elif self._in_packet['remaining_length'] != 2:
- return MQTTErrorCode.MQTT_ERR_PROTOCOL
-
- mid, = struct.unpack("!H", self._in_packet['packet'][:2])
- if self._protocol == MQTTv5:
- packet = self._in_packet['packet'][2:]
- properties = Properties(UNSUBACK >> 4)
- props, props_len = properties.unpack(packet)
- reasoncodes_list = [
- ReasonCode(UNSUBACK >> 4, identifier=c)
- for c in packet[props_len:]
- ]
- else:
- reasoncodes_list = []
- properties = Properties(UNSUBACK >> 4)
-
- self._easy_log(MQTT_LOG_DEBUG, "Received UNSUBACK (Mid: %d)", mid)
- with self._callback_mutex:
- on_unsubscribe = self.on_unsubscribe
-
- if on_unsubscribe:
- with self._in_callback_mutex:
- try:
- if self._callback_api_version == CallbackAPIVersion.VERSION1:
- if self._protocol == MQTTv5:
- on_unsubscribe = cast(CallbackOnUnsubscribe_v1_mqtt5, on_unsubscribe)
-
- reasoncodes: ReasonCode | list[ReasonCode] = reasoncodes_list
- if len(reasoncodes_list) == 1:
- reasoncodes = reasoncodes_list[0]
-
- on_unsubscribe(
- self, self._userdata, mid, properties, reasoncodes)
- else:
- on_unsubscribe = cast(CallbackOnUnsubscribe_v1_mqtt3, on_unsubscribe)
-
- on_unsubscribe(self, self._userdata, mid)
- elif self._callback_api_version == CallbackAPIVersion.VERSION2:
- on_unsubscribe = cast(CallbackOnUnsubscribe_v2, on_unsubscribe)
-
- if properties is None:
- properties = Properties(PacketTypes.CONNACK)
-
- on_unsubscribe(
- self,
- self._userdata,
- mid,
- reasoncodes_list,
- properties,
- )
- else:
- raise RuntimeError("Unsupported callback API version")
- except Exception as err:
- self._easy_log(
- MQTT_LOG_ERR, 'Caught exception in on_unsubscribe: %s', err)
- if not self.suppress_exceptions:
- raise
-
- return MQTTErrorCode.MQTT_ERR_SUCCESS
-
- def _do_on_disconnect(
- self,
- packet_from_broker: bool,
- v1_rc: MQTTErrorCode,
- reason: ReasonCode | None = None,
- properties: Properties | None = None,
- ) -> None:
- with self._callback_mutex:
- on_disconnect = self.on_disconnect
-
- if on_disconnect:
- with self._in_callback_mutex:
- try:
- if self._callback_api_version == CallbackAPIVersion.VERSION1:
- if self._protocol == MQTTv5:
- on_disconnect = cast(CallbackOnDisconnect_v1_mqtt5, on_disconnect)
-
- if packet_from_broker:
- on_disconnect(self, self._userdata, reason, properties)
- else:
- on_disconnect(self, self._userdata, v1_rc, None)
- else:
- on_disconnect = cast(CallbackOnDisconnect_v1_mqtt3, on_disconnect)
-
- on_disconnect(self, self._userdata, v1_rc)
- elif self._callback_api_version == CallbackAPIVersion.VERSION2:
- on_disconnect = cast(CallbackOnDisconnect_v2, on_disconnect)
-
- disconnect_flags = DisconnectFlags(
- is_disconnect_packet_from_server=packet_from_broker
- )
-
- if reason is None:
- reason = convert_disconnect_error_code_to_reason_code(v1_rc)
-
- if properties is None:
- properties = Properties(PacketTypes.DISCONNECT)
-
- on_disconnect(
- self,
- self._userdata,
- disconnect_flags,
- reason,
- properties,
- )
- else:
- raise RuntimeError("Unsupported callback API version")
- except Exception as err:
- self._easy_log(
- MQTT_LOG_ERR, 'Caught exception in on_disconnect: %s', err)
- if not self.suppress_exceptions:
- raise
-
- def _do_on_publish(self, mid: int, reason_code: ReasonCode, properties: Properties) -> MQTTErrorCode:
- with self._callback_mutex:
- on_publish = self.on_publish
-
- if on_publish:
- with self._in_callback_mutex:
- try:
- if self._callback_api_version == CallbackAPIVersion.VERSION1:
- on_publish = cast(CallbackOnPublish_v1, on_publish)
-
- on_publish(self, self._userdata, mid)
- elif self._callback_api_version == CallbackAPIVersion.VERSION2:
- on_publish = cast(CallbackOnPublish_v2, on_publish)
-
- on_publish(
- self,
- self._userdata,
- mid,
- reason_code,
- properties,
- )
- else:
- raise RuntimeError("Unsupported callback API version")
- except Exception as err:
- self._easy_log(
- MQTT_LOG_ERR, 'Caught exception in on_publish: %s', err)
- if not self.suppress_exceptions:
- raise
-
- msg = self._out_messages.pop(mid)
- msg.info._set_as_published()
- if msg.qos > 0:
- self._inflight_messages -= 1
- if self._max_inflight_messages > 0:
- rc = self._update_inflight()
- if rc != MQTTErrorCode.MQTT_ERR_SUCCESS:
- return rc
- return MQTTErrorCode.MQTT_ERR_SUCCESS
-
- def _handle_pubackcomp(
- self, cmd: Literal['PUBACK'] | Literal['PUBCOMP']
- ) -> MQTTErrorCode:
- if self._protocol == MQTTv5:
- if self._in_packet['remaining_length'] < 2:
- return MQTTErrorCode.MQTT_ERR_PROTOCOL
- elif self._in_packet['remaining_length'] != 2:
- return MQTTErrorCode.MQTT_ERR_PROTOCOL
-
- packet_type_enum = PUBACK if cmd == "PUBACK" else PUBCOMP
- packet_type = packet_type_enum.value >> 4
- mid, = struct.unpack("!H", self._in_packet['packet'][:2])
- reasonCode = ReasonCode(packet_type)
- properties = Properties(packet_type)
- if self._protocol == MQTTv5:
- if self._in_packet['remaining_length'] > 2:
- reasonCode.unpack(self._in_packet['packet'][2:])
- if self._in_packet['remaining_length'] > 3:
- props, props_len = properties.unpack(
- self._in_packet['packet'][3:])
- self._easy_log(MQTT_LOG_DEBUG, "Received %s (Mid: %d)", cmd, mid)
-
- with self._out_message_mutex:
- if mid in self._out_messages:
- # Only inform the client the message has been sent once.
- rc = self._do_on_publish(mid, reasonCode, properties)
- return rc
-
- return MQTTErrorCode.MQTT_ERR_SUCCESS
-
- def _handle_on_message(self, message: MQTTMessage) -> None:
-
- try:
- topic = message.topic
- except UnicodeDecodeError:
- topic = None
-
- on_message_callbacks = []
- with self._callback_mutex:
- if topic is not None:
- on_message_callbacks = list(self._on_message_filtered.iter_match(message.topic))
-
- if len(on_message_callbacks) == 0:
- on_message = self.on_message
- else:
- on_message = None
-
- for callback in on_message_callbacks:
- with self._in_callback_mutex:
- try:
- callback(self, self._userdata, message)
- except Exception as err:
- self._easy_log(
- MQTT_LOG_ERR,
- 'Caught exception in user defined callback function %s: %s',
- callback.__name__,
- err
- )
- if not self.suppress_exceptions:
- raise
-
- if on_message:
- with self._in_callback_mutex:
- try:
- on_message(self, self._userdata, message)
- except Exception as err:
- self._easy_log(
- MQTT_LOG_ERR, 'Caught exception in on_message: %s', err)
- if not self.suppress_exceptions:
- raise
-
-
- def _handle_on_connect_fail(self) -> None:
- with self._callback_mutex:
- on_connect_fail = self.on_connect_fail
-
- if on_connect_fail:
- with self._in_callback_mutex:
- try:
- on_connect_fail(self, self._userdata)
- except Exception as err:
- self._easy_log(
- MQTT_LOG_ERR, 'Caught exception in on_connect_fail: %s', err)
-
- def _thread_main(self) -> None:
- try:
- self.loop_forever(retry_first_connection=True)
- finally:
- self._thread = None
-
- def _reconnect_wait(self) -> None:
- # See reconnect_delay_set for details
- now = time_func()
- with self._reconnect_delay_mutex:
- if self._reconnect_delay is None:
- self._reconnect_delay = self._reconnect_min_delay
- else:
- self._reconnect_delay = min(
- self._reconnect_delay * 2,
- self._reconnect_max_delay,
- )
-
- target_time = now + self._reconnect_delay
-
- remaining = target_time - now
- while (self._state not in (_ConnectionState.MQTT_CS_DISCONNECTING, _ConnectionState.MQTT_CS_DISCONNECTED)
- and not self._thread_terminate
- and remaining > 0):
-
- time.sleep(min(remaining, 1))
- remaining = target_time - time_func()
-
- @staticmethod
- def _proxy_is_valid(p) -> bool: # type: ignore[no-untyped-def]
- def check(t, a) -> bool: # type: ignore[no-untyped-def]
- return (socks is not None and
- t in {socks.HTTP, socks.SOCKS4, socks.SOCKS5} and a)
-
- if isinstance(p, dict):
- return check(p.get("proxy_type"), p.get("proxy_addr"))
- elif isinstance(p, (list, tuple)):
- return len(p) == 6 and check(p[0], p[1])
- else:
- return False
-
- def _get_proxy(self) -> dict[str, Any] | None:
- if socks is None:
- return None
-
- # First, check if the user explicitly passed us a proxy to use
- if self._proxy_is_valid(self._proxy):
- return self._proxy
-
- # Next, check for an mqtt_proxy environment variable as long as the host
- # we're trying to connect to isn't listed under the no_proxy environment
- # variable (matches built-in module urllib's behavior)
- if not (hasattr(urllib.request, "proxy_bypass") and
- urllib.request.proxy_bypass(self._host)):
- env_proxies = urllib.request.getproxies()
- if "mqtt" in env_proxies:
- parts = urllib.parse.urlparse(env_proxies["mqtt"])
- if parts.scheme == "http":
- proxy = {
- "proxy_type": socks.HTTP,
- "proxy_addr": parts.hostname,
- "proxy_port": parts.port
- }
- return proxy
- elif parts.scheme == "socks":
- proxy = {
- "proxy_type": socks.SOCKS5,
- "proxy_addr": parts.hostname,
- "proxy_port": parts.port
- }
- return proxy
-
- # Finally, check if the user has monkeypatched the PySocks library with
- # a default proxy
- socks_default = socks.get_default_proxy()
- if self._proxy_is_valid(socks_default):
- proxy_keys = ("proxy_type", "proxy_addr", "proxy_port",
- "proxy_rdns", "proxy_username", "proxy_password")
- return dict(zip(proxy_keys, socks_default))
-
- # If we didn't find a proxy through any of the above methods, return
- # None to indicate that the connection should be handled normally
- return None
-
- def _create_socket(self) -> SocketLike:
- if self._transport == "unix":
- sock = self._create_unix_socket_connection()
- else:
- sock = self._create_socket_connection()
-
- if self._ssl:
- sock = self._ssl_wrap_socket(sock)
-
- if self._transport == "websockets":
- sock.settimeout(self._keepalive)
- return _WebsocketWrapper(
- socket=sock,
- host=self._host,
- port=self._port,
- is_ssl=self._ssl,
- path=self._websocket_path,
- extra_headers=self._websocket_extra_headers,
- )
-
- return sock
-
- def _create_unix_socket_connection(self) -> _socket.socket:
- unix_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
- unix_socket.connect(self._host)
- return unix_socket
-
- def _create_socket_connection(self) -> _socket.socket:
- proxy = self._get_proxy()
- addr = (self._host, self._port)
- source = (self._bind_address, self._bind_port)
-
- if proxy:
- return socks.create_connection(addr, timeout=self._connect_timeout, source_address=source, **proxy)
- else:
- return socket.create_connection(addr, timeout=self._connect_timeout, source_address=source)
-
- def _ssl_wrap_socket(self, tcp_sock: _socket.socket) -> ssl.SSLSocket:
- if self._ssl_context is None:
- raise ValueError(
- "Impossible condition. _ssl_context should never be None if _ssl is True"
- )
-
- verify_host = not self._tls_insecure
- try:
- # Try with server_hostname, even it's not supported in certain scenarios
- ssl_sock = self._ssl_context.wrap_socket(
- tcp_sock,
- server_hostname=self._host,
- do_handshake_on_connect=False,
- )
- except ssl.CertificateError:
- # CertificateError is derived from ValueError
- raise
- except ValueError:
- # Python version requires SNI in order to handle server_hostname, but SNI is not available
- ssl_sock = self._ssl_context.wrap_socket(
- tcp_sock,
- do_handshake_on_connect=False,
- )
- else:
- # If SSL context has already checked hostname, then don't need to do it again
- if getattr(self._ssl_context, 'check_hostname', False): # type: ignore
- verify_host = False
-
- ssl_sock.settimeout(self._keepalive)
- ssl_sock.do_handshake()
-
- if verify_host:
- # TODO: this type error is a true error:
- # error: Module has no attribute "match_hostname" [attr-defined]
- # Python 3.12 no longer have this method.
- ssl.match_hostname(ssl_sock.getpeercert(), self._host) # type: ignore
-
- return ssl_sock
-
-class _WebsocketWrapper:
- OPCODE_CONTINUATION = 0x0
- OPCODE_TEXT = 0x1
- OPCODE_BINARY = 0x2
- OPCODE_CONNCLOSE = 0x8
- OPCODE_PING = 0x9
- OPCODE_PONG = 0xa
-
- def __init__(
- self,
- socket: socket.socket | ssl.SSLSocket,
- host: str,
- port: int,
- is_ssl: bool,
- path: str,
- extra_headers: WebSocketHeaders | None,
- ):
- self.connected = False
-
- self._ssl = is_ssl
- self._host = host
- self._port = port
- self._socket = socket
- self._path = path
-
- self._sendbuffer = bytearray()
- self._readbuffer = bytearray()
-
- self._requested_size = 0
- self._payload_head = 0
- self._readbuffer_head = 0
-
- self._do_handshake(extra_headers)
-
- def __del__(self) -> None:
- self._sendbuffer = bytearray()
- self._readbuffer = bytearray()
-
- def _do_handshake(self, extra_headers: WebSocketHeaders | None) -> None:
-
- sec_websocket_key = uuid.uuid4().bytes
- sec_websocket_key = base64.b64encode(sec_websocket_key)
-
- if self._ssl:
- default_port = 443
- http_schema = "https"
- else:
- default_port = 80
- http_schema = "http"
-
- if default_port == self._port:
- host_port = f"{self._host}"
- else:
- host_port = f"{self._host}:{self._port}"
-
- websocket_headers = {
- "Host": host_port,
- "Upgrade": "websocket",
- "Connection": "Upgrade",
- "Origin": f"{http_schema}://{host_port}",
- "Sec-WebSocket-Key": sec_websocket_key.decode("utf8"),
- "Sec-Websocket-Version": "13",
- "Sec-Websocket-Protocol": "mqtt",
- }
-
- # This is checked in ws_set_options so it will either be None, a
- # dictionary, or a callable
- if isinstance(extra_headers, dict):
- websocket_headers.update(extra_headers)
- elif callable(extra_headers):
- websocket_headers = extra_headers(websocket_headers)
-
- header = "\r\n".join([
- f"GET {self._path} HTTP/1.1",
- "\r\n".join(f"{i}: {j}" for i, j in websocket_headers.items()),
- "\r\n",
- ]).encode("utf8")
-
- self._socket.send(header)
-
- has_secret = False
- has_upgrade = False
-
- while True:
- # read HTTP response header as lines
- try:
- byte = self._socket.recv(1)
- except ConnectionResetError:
- byte = b""
-
- self._readbuffer.extend(byte)
-
- # line end
- if byte == b"\n":
- if len(self._readbuffer) > 2:
- # check upgrade
- if b"connection" in str(self._readbuffer).lower().encode('utf-8'):
- if b"upgrade" not in str(self._readbuffer).lower().encode('utf-8'):
- raise WebsocketConnectionError(
- "WebSocket handshake error, connection not upgraded")
- else:
- has_upgrade = True
-
- # check key hash
- if b"sec-websocket-accept" in str(self._readbuffer).lower().encode('utf-8'):
- GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
-
- server_hash_str = self._readbuffer.decode(
- 'utf-8').split(": ", 1)[1]
- server_hash = server_hash_str.strip().encode('utf-8')
-
- client_hash_key = sec_websocket_key.decode('utf-8') + GUID
- # Use of SHA-1 is OK here; it's according to the Websocket spec.
- client_hash_digest = hashlib.sha1(client_hash_key.encode('utf-8')) # noqa: S324
- client_hash = base64.b64encode(client_hash_digest.digest())
-
- if server_hash != client_hash:
- raise WebsocketConnectionError(
- "WebSocket handshake error, invalid secret key")
- else:
- has_secret = True
- else:
- # ending linebreak
- break
-
- # reset linebuffer
- self._readbuffer = bytearray()
-
- # connection reset
- elif not byte:
- raise WebsocketConnectionError("WebSocket handshake error")
-
- if not has_upgrade or not has_secret:
- raise WebsocketConnectionError("WebSocket handshake error")
-
- self._readbuffer = bytearray()
- self.connected = True
-
- def _create_frame(
- self, opcode: int, data: bytearray, do_masking: int = 1
- ) -> bytearray:
- header = bytearray()
- length = len(data)
-
- mask_key = bytearray(os.urandom(4))
- mask_flag = do_masking
-
- # 1 << 7 is the final flag, we don't send continuated data
- header.append(1 << 7 | opcode)
-
- if length < 126:
- header.append(mask_flag << 7 | length)
-
- elif length < 65536:
- header.append(mask_flag << 7 | 126)
- header += struct.pack("!H", length)
-
- elif length < 0x8000000000000001:
- header.append(mask_flag << 7 | 127)
- header += struct.pack("!Q", length)
-
- else:
- raise ValueError("Maximum payload size is 2^63")
-
- if mask_flag == 1:
- for index in range(length):
- data[index] ^= mask_key[index % 4]
- data = mask_key + data
-
- return header + data
-
- def _buffered_read(self, length: int) -> bytearray:
-
- # try to recv and store needed bytes
- wanted_bytes = length - (len(self._readbuffer) - self._readbuffer_head)
- if wanted_bytes > 0:
-
- data = self._socket.recv(wanted_bytes)
-
- if not data:
- raise ConnectionAbortedError
- else:
- self._readbuffer.extend(data)
-
- if len(data) < wanted_bytes:
- raise BlockingIOError
-
- self._readbuffer_head += length
- return self._readbuffer[self._readbuffer_head - length:self._readbuffer_head]
-
- def _recv_impl(self, length: int) -> bytes:
-
- # try to decode websocket payload part from data
- try:
-
- self._readbuffer_head = 0
-
- result = b""
-
- chunk_startindex = self._payload_head
- chunk_endindex = self._payload_head + length
-
- header1 = self._buffered_read(1)
- header2 = self._buffered_read(1)
-
- opcode = (header1[0] & 0x0f)
- maskbit = (header2[0] & 0x80) == 0x80
- lengthbits = (header2[0] & 0x7f)
- payload_length = lengthbits
- mask_key = None
-
- # read length
- if lengthbits == 0x7e:
-
- value = self._buffered_read(2)
- payload_length, = struct.unpack("!H", value)
-
- elif lengthbits == 0x7f:
-
- value = self._buffered_read(8)
- payload_length, = struct.unpack("!Q", value)
-
- # read mask
- if maskbit:
- mask_key = self._buffered_read(4)
-
- # if frame payload is shorter than the requested data, read only the possible part
- readindex = chunk_endindex
- if payload_length < readindex:
- readindex = payload_length
-
- if readindex > 0:
- # get payload chunk
- payload = self._buffered_read(readindex)
-
- # unmask only the needed part
- if mask_key is not None:
- for index in range(chunk_startindex, readindex):
- payload[index] ^= mask_key[index % 4]
-
- result = payload[chunk_startindex:readindex]
- self._payload_head = readindex
- else:
- payload = bytearray()
-
- # check if full frame arrived and reset readbuffer and payloadhead if needed
- if readindex == payload_length:
- self._readbuffer = bytearray()
- self._payload_head = 0
-
- # respond to non-binary opcodes, their arrival is not guaranteed because of non-blocking sockets
- if opcode == _WebsocketWrapper.OPCODE_CONNCLOSE:
- frame = self._create_frame(
- _WebsocketWrapper.OPCODE_CONNCLOSE, payload, 0)
- self._socket.send(frame)
-
- if opcode == _WebsocketWrapper.OPCODE_PING:
- frame = self._create_frame(
- _WebsocketWrapper.OPCODE_PONG, payload, 0)
- self._socket.send(frame)
-
- # This isn't *proper* handling of continuation frames, but given
- # that we only support binary frames, it is *probably* good enough.
- if (opcode == _WebsocketWrapper.OPCODE_BINARY or opcode == _WebsocketWrapper.OPCODE_CONTINUATION) \
- and payload_length > 0:
- return result
- else:
- raise BlockingIOError
-
- except ConnectionError:
- self.connected = False
- return b''
-
- def _send_impl(self, data: bytes) -> int:
-
- # if previous frame was sent successfully
- if len(self._sendbuffer) == 0:
- # create websocket frame
- frame = self._create_frame(
- _WebsocketWrapper.OPCODE_BINARY, bytearray(data))
- self._sendbuffer.extend(frame)
- self._requested_size = len(data)
-
- # try to write out as much as possible
- length = self._socket.send(self._sendbuffer)
-
- self._sendbuffer = self._sendbuffer[length:]
-
- if len(self._sendbuffer) == 0:
- # buffer sent out completely, return with payload's size
- return self._requested_size
- else:
- # couldn't send whole data, request the same data again with 0 as sent length
- return 0
-
- def recv(self, length: int) -> bytes:
- return self._recv_impl(length)
-
- def read(self, length: int) -> bytes:
- return self._recv_impl(length)
-
- def send(self, data: bytes) -> int:
- return self._send_impl(data)
-
- def write(self, data: bytes) -> int:
- return self._send_impl(data)
-
- def close(self) -> None:
- self._socket.close()
-
- def fileno(self) -> int:
- return self._socket.fileno()
-
- def pending(self) -> int:
- # Fix for bug #131: a SSL socket may still have data available
- # for reading without select() being aware of it.
- if self._ssl:
- return self._socket.pending() # type: ignore[union-attr]
- else:
- # normal socket rely only on select()
- return 0
-
- def setblocking(self, flag: bool) -> None:
- self._socket.setblocking(flag)
diff --git a/sbapp/mqtt/enums.py b/sbapp/mqtt/enums.py
deleted file mode 100644
index 5428769f..00000000
--- a/sbapp/mqtt/enums.py
+++ /dev/null
@@ -1,113 +0,0 @@
-import enum
-
-
-class MQTTErrorCode(enum.IntEnum):
- MQTT_ERR_AGAIN = -1
- MQTT_ERR_SUCCESS = 0
- MQTT_ERR_NOMEM = 1
- MQTT_ERR_PROTOCOL = 2
- MQTT_ERR_INVAL = 3
- MQTT_ERR_NO_CONN = 4
- MQTT_ERR_CONN_REFUSED = 5
- MQTT_ERR_NOT_FOUND = 6
- MQTT_ERR_CONN_LOST = 7
- MQTT_ERR_TLS = 8
- MQTT_ERR_PAYLOAD_SIZE = 9
- MQTT_ERR_NOT_SUPPORTED = 10
- MQTT_ERR_AUTH = 11
- MQTT_ERR_ACL_DENIED = 12
- MQTT_ERR_UNKNOWN = 13
- MQTT_ERR_ERRNO = 14
- MQTT_ERR_QUEUE_SIZE = 15
- MQTT_ERR_KEEPALIVE = 16
-
-
-class MQTTProtocolVersion(enum.IntEnum):
- MQTTv31 = 3
- MQTTv311 = 4
- MQTTv5 = 5
-
-
-class CallbackAPIVersion(enum.Enum):
- """Defined the arguments passed to all user-callback.
-
- See each callbacks for details: `on_connect`, `on_connect_fail`, `on_disconnect`, `on_message`, `on_publish`,
- `on_subscribe`, `on_unsubscribe`, `on_log`, `on_socket_open`, `on_socket_close`,
- `on_socket_register_write`, `on_socket_unregister_write`
- """
- VERSION1 = 1
- """The version used with paho-mqtt 1.x before introducing CallbackAPIVersion.
-
- This version had different arguments depending if MQTTv5 or MQTTv3 was used. `Properties` & `ReasonCode` were missing
- on some callback (apply only to MQTTv5).
-
- This version is deprecated and will be removed in version 3.0.
- """
- VERSION2 = 2
- """ This version fix some of the shortcoming of previous version.
-
- Callback have the same signature if using MQTTv5 or MQTTv3. `ReasonCode` are used in MQTTv3.
- """
-
-
-class MessageType(enum.IntEnum):
- CONNECT = 0x10
- CONNACK = 0x20
- PUBLISH = 0x30
- PUBACK = 0x40
- PUBREC = 0x50
- PUBREL = 0x60
- PUBCOMP = 0x70
- SUBSCRIBE = 0x80
- SUBACK = 0x90
- UNSUBSCRIBE = 0xA0
- UNSUBACK = 0xB0
- PINGREQ = 0xC0
- PINGRESP = 0xD0
- DISCONNECT = 0xE0
- AUTH = 0xF0
-
-
-class LogLevel(enum.IntEnum):
- MQTT_LOG_INFO = 0x01
- MQTT_LOG_NOTICE = 0x02
- MQTT_LOG_WARNING = 0x04
- MQTT_LOG_ERR = 0x08
- MQTT_LOG_DEBUG = 0x10
-
-
-class ConnackCode(enum.IntEnum):
- CONNACK_ACCEPTED = 0
- CONNACK_REFUSED_PROTOCOL_VERSION = 1
- CONNACK_REFUSED_IDENTIFIER_REJECTED = 2
- CONNACK_REFUSED_SERVER_UNAVAILABLE = 3
- CONNACK_REFUSED_BAD_USERNAME_PASSWORD = 4
- CONNACK_REFUSED_NOT_AUTHORIZED = 5
-
-
-class _ConnectionState(enum.Enum):
- MQTT_CS_NEW = enum.auto()
- MQTT_CS_CONNECT_ASYNC = enum.auto()
- MQTT_CS_CONNECTING = enum.auto()
- MQTT_CS_CONNECTED = enum.auto()
- MQTT_CS_CONNECTION_LOST = enum.auto()
- MQTT_CS_DISCONNECTING = enum.auto()
- MQTT_CS_DISCONNECTED = enum.auto()
-
-
-class MessageState(enum.IntEnum):
- MQTT_MS_INVALID = 0
- MQTT_MS_PUBLISH = 1
- MQTT_MS_WAIT_FOR_PUBACK = 2
- MQTT_MS_WAIT_FOR_PUBREC = 3
- MQTT_MS_RESEND_PUBREL = 4
- MQTT_MS_WAIT_FOR_PUBREL = 5
- MQTT_MS_RESEND_PUBCOMP = 6
- MQTT_MS_WAIT_FOR_PUBCOMP = 7
- MQTT_MS_SEND_PUBREC = 8
- MQTT_MS_QUEUED = 9
-
-
-class PahoClientMode(enum.IntEnum):
- MQTT_CLIENT = 0
- MQTT_BRIDGE = 1
diff --git a/sbapp/mqtt/matcher.py b/sbapp/mqtt/matcher.py
deleted file mode 100644
index b73c13ac..00000000
--- a/sbapp/mqtt/matcher.py
+++ /dev/null
@@ -1,78 +0,0 @@
-class MQTTMatcher:
- """Intended to manage topic filters including wildcards.
-
- Internally, MQTTMatcher use a prefix tree (trie) to store
- values associated with filters, and has an iter_match()
- method to iterate efficiently over all filters that match
- some topic name."""
-
- class Node:
- __slots__ = '_children', '_content'
-
- def __init__(self):
- self._children = {}
- self._content = None
-
- def __init__(self):
- self._root = self.Node()
-
- def __setitem__(self, key, value):
- """Add a topic filter :key to the prefix tree
- and associate it to :value"""
- node = self._root
- for sym in key.split('/'):
- node = node._children.setdefault(sym, self.Node())
- node._content = value
-
- def __getitem__(self, key):
- """Retrieve the value associated with some topic filter :key"""
- try:
- node = self._root
- for sym in key.split('/'):
- node = node._children[sym]
- if node._content is None:
- raise KeyError(key)
- return node._content
- except KeyError as ke:
- raise KeyError(key) from ke
-
- def __delitem__(self, key):
- """Delete the value associated with some topic filter :key"""
- lst = []
- try:
- parent, node = None, self._root
- for k in key.split('/'):
- parent, node = node, node._children[k]
- lst.append((parent, k, node))
- # TODO
- node._content = None
- except KeyError as ke:
- raise KeyError(key) from ke
- else: # cleanup
- for parent, k, node in reversed(lst):
- if node._children or node._content is not None:
- break
- del parent._children[k]
-
- def iter_match(self, topic):
- """Return an iterator on all values associated with filters
- that match the :topic"""
- lst = topic.split('/')
- normal = not topic.startswith('$')
- def rec(node, i=0):
- if i == len(lst):
- if node._content is not None:
- yield node._content
- else:
- part = lst[i]
- if part in node._children:
- for content in rec(node._children[part], i + 1):
- yield content
- if '+' in node._children and (normal or i > 0):
- for content in rec(node._children['+'], i + 1):
- yield content
- if '#' in node._children and (normal or i > 0):
- content = node._children['#']._content
- if content is not None:
- yield content
- return rec(self._root)
diff --git a/sbapp/mqtt/packettypes.py b/sbapp/mqtt/packettypes.py
deleted file mode 100644
index d2051490..00000000
--- a/sbapp/mqtt/packettypes.py
+++ /dev/null
@@ -1,43 +0,0 @@
-"""
-*******************************************************************
- Copyright (c) 2017, 2019 IBM Corp.
-
- All rights reserved. This program and the accompanying materials
- are made available under the terms of the Eclipse Public License v2.0
- and Eclipse Distribution License v1.0 which accompany this distribution.
-
- The Eclipse Public License is available at
- http://www.eclipse.org/legal/epl-v20.html
- and the Eclipse Distribution License is available at
- http://www.eclipse.org/org/documents/edl-v10.php.
-
- Contributors:
- Ian Craggs - initial implementation and/or documentation
-*******************************************************************
-"""
-
-
-class PacketTypes:
-
- """
- Packet types class. Includes the AUTH packet for MQTT v5.0.
-
- Holds constants for each packet type such as PacketTypes.PUBLISH
- and packet name strings: PacketTypes.Names[PacketTypes.PUBLISH].
-
- """
-
- indexes = range(1, 16)
-
- # Packet types
- CONNECT, CONNACK, PUBLISH, PUBACK, PUBREC, PUBREL, \
- PUBCOMP, SUBSCRIBE, SUBACK, UNSUBSCRIBE, UNSUBACK, \
- PINGREQ, PINGRESP, DISCONNECT, AUTH = indexes
-
- # Dummy packet type for properties use - will delay only applies to will
- WILLMESSAGE = 99
-
- Names = ( "reserved", \
- "Connect", "Connack", "Publish", "Puback", "Pubrec", "Pubrel", \
- "Pubcomp", "Subscribe", "Suback", "Unsubscribe", "Unsuback", \
- "Pingreq", "Pingresp", "Disconnect", "Auth")
diff --git a/sbapp/mqtt/properties.py b/sbapp/mqtt/properties.py
deleted file mode 100644
index f307b865..00000000
--- a/sbapp/mqtt/properties.py
+++ /dev/null
@@ -1,421 +0,0 @@
-# *******************************************************************
-# Copyright (c) 2017, 2019 IBM Corp.
-#
-# All rights reserved. This program and the accompanying materials
-# are made available under the terms of the Eclipse Public License v2.0
-# and Eclipse Distribution License v1.0 which accompany this distribution.
-#
-# The Eclipse Public License is available at
-# http://www.eclipse.org/legal/epl-v20.html
-# and the Eclipse Distribution License is available at
-# http://www.eclipse.org/org/documents/edl-v10.php.
-#
-# Contributors:
-# Ian Craggs - initial implementation and/or documentation
-# *******************************************************************
-
-import struct
-
-from .packettypes import PacketTypes
-
-
-class MQTTException(Exception):
- pass
-
-
-class MalformedPacket(MQTTException):
- pass
-
-
-def writeInt16(length):
- # serialize a 16 bit integer to network format
- return bytearray(struct.pack("!H", length))
-
-
-def readInt16(buf):
- # deserialize a 16 bit integer from network format
- return struct.unpack("!H", buf[:2])[0]
-
-
-def writeInt32(length):
- # serialize a 32 bit integer to network format
- return bytearray(struct.pack("!L", length))
-
-
-def readInt32(buf):
- # deserialize a 32 bit integer from network format
- return struct.unpack("!L", buf[:4])[0]
-
-
-def writeUTF(data):
- # data could be a string, or bytes. If string, encode into bytes with utf-8
- if not isinstance(data, bytes):
- data = bytes(data, "utf-8")
- return writeInt16(len(data)) + data
-
-
-def readUTF(buffer, maxlen):
- if maxlen >= 2:
- length = readInt16(buffer)
- else:
- raise MalformedPacket("Not enough data to read string length")
- maxlen -= 2
- if length > maxlen:
- raise MalformedPacket("Length delimited string too long")
- buf = buffer[2:2+length].decode("utf-8")
- # look for chars which are invalid for MQTT
- for c in buf: # look for D800-DFFF in the UTF string
- ord_c = ord(c)
- if ord_c >= 0xD800 and ord_c <= 0xDFFF:
- raise MalformedPacket("[MQTT-1.5.4-1] D800-DFFF found in UTF-8 data")
- if ord_c == 0x00: # look for null in the UTF string
- raise MalformedPacket("[MQTT-1.5.4-2] Null found in UTF-8 data")
- if ord_c == 0xFEFF:
- raise MalformedPacket("[MQTT-1.5.4-3] U+FEFF in UTF-8 data")
- return buf, length+2
-
-
-def writeBytes(buffer):
- return writeInt16(len(buffer)) + buffer
-
-
-def readBytes(buffer):
- length = readInt16(buffer)
- return buffer[2:2+length], length+2
-
-
-class VariableByteIntegers: # Variable Byte Integer
- """
- MQTT variable byte integer helper class. Used
- in several places in MQTT v5.0 properties.
-
- """
-
- @staticmethod
- def encode(x):
- """
- Convert an integer 0 <= x <= 268435455 into multi-byte format.
- Returns the buffer converted from the integer.
- """
- if not 0 <= x <= 268435455:
- raise ValueError(f"Value {x!r} must be in range 0-268435455")
- buffer = b''
- while 1:
- digit = x % 128
- x //= 128
- if x > 0:
- digit |= 0x80
- buffer += bytes([digit])
- if x == 0:
- break
- return buffer
-
- @staticmethod
- def decode(buffer):
- """
- Get the value of a multi-byte integer from a buffer
- Return the value, and the number of bytes used.
-
- [MQTT-1.5.5-1] the encoded value MUST use the minimum number of bytes necessary to represent the value
- """
- multiplier = 1
- value = 0
- bytes = 0
- while 1:
- bytes += 1
- digit = buffer[0]
- buffer = buffer[1:]
- value += (digit & 127) * multiplier
- if digit & 128 == 0:
- break
- multiplier *= 128
- return (value, bytes)
-
-
-class Properties:
- """MQTT v5.0 properties class.
-
- See Properties.names for a list of accepted property names along with their numeric values.
-
- See Properties.properties for the data type of each property.
-
- Example of use::
-
- publish_properties = Properties(PacketTypes.PUBLISH)
- publish_properties.UserProperty = ("a", "2")
- publish_properties.UserProperty = ("c", "3")
-
- First the object is created with packet type as argument, no properties will be present at
- this point. Then properties are added as attributes, the name of which is the string property
- name without the spaces.
-
- """
-
- def __init__(self, packetType):
- self.packetType = packetType
- self.types = ["Byte", "Two Byte Integer", "Four Byte Integer", "Variable Byte Integer",
- "Binary Data", "UTF-8 Encoded String", "UTF-8 String Pair"]
-
- self.names = {
- "Payload Format Indicator": 1,
- "Message Expiry Interval": 2,
- "Content Type": 3,
- "Response Topic": 8,
- "Correlation Data": 9,
- "Subscription Identifier": 11,
- "Session Expiry Interval": 17,
- "Assigned Client Identifier": 18,
- "Server Keep Alive": 19,
- "Authentication Method": 21,
- "Authentication Data": 22,
- "Request Problem Information": 23,
- "Will Delay Interval": 24,
- "Request Response Information": 25,
- "Response Information": 26,
- "Server Reference": 28,
- "Reason String": 31,
- "Receive Maximum": 33,
- "Topic Alias Maximum": 34,
- "Topic Alias": 35,
- "Maximum QoS": 36,
- "Retain Available": 37,
- "User Property": 38,
- "Maximum Packet Size": 39,
- "Wildcard Subscription Available": 40,
- "Subscription Identifier Available": 41,
- "Shared Subscription Available": 42
- }
-
- self.properties = {
- # id: type, packets
- # payload format indicator
- 1: (self.types.index("Byte"), [PacketTypes.PUBLISH, PacketTypes.WILLMESSAGE]),
- 2: (self.types.index("Four Byte Integer"), [PacketTypes.PUBLISH, PacketTypes.WILLMESSAGE]),
- 3: (self.types.index("UTF-8 Encoded String"), [PacketTypes.PUBLISH, PacketTypes.WILLMESSAGE]),
- 8: (self.types.index("UTF-8 Encoded String"), [PacketTypes.PUBLISH, PacketTypes.WILLMESSAGE]),
- 9: (self.types.index("Binary Data"), [PacketTypes.PUBLISH, PacketTypes.WILLMESSAGE]),
- 11: (self.types.index("Variable Byte Integer"),
- [PacketTypes.PUBLISH, PacketTypes.SUBSCRIBE]),
- 17: (self.types.index("Four Byte Integer"),
- [PacketTypes.CONNECT, PacketTypes.CONNACK, PacketTypes.DISCONNECT]),
- 18: (self.types.index("UTF-8 Encoded String"), [PacketTypes.CONNACK]),
- 19: (self.types.index("Two Byte Integer"), [PacketTypes.CONNACK]),
- 21: (self.types.index("UTF-8 Encoded String"),
- [PacketTypes.CONNECT, PacketTypes.CONNACK, PacketTypes.AUTH]),
- 22: (self.types.index("Binary Data"),
- [PacketTypes.CONNECT, PacketTypes.CONNACK, PacketTypes.AUTH]),
- 23: (self.types.index("Byte"),
- [PacketTypes.CONNECT]),
- 24: (self.types.index("Four Byte Integer"), [PacketTypes.WILLMESSAGE]),
- 25: (self.types.index("Byte"), [PacketTypes.CONNECT]),
- 26: (self.types.index("UTF-8 Encoded String"), [PacketTypes.CONNACK]),
- 28: (self.types.index("UTF-8 Encoded String"),
- [PacketTypes.CONNACK, PacketTypes.DISCONNECT]),
- 31: (self.types.index("UTF-8 Encoded String"),
- [PacketTypes.CONNACK, PacketTypes.PUBACK, PacketTypes.PUBREC,
- PacketTypes.PUBREL, PacketTypes.PUBCOMP, PacketTypes.SUBACK,
- PacketTypes.UNSUBACK, PacketTypes.DISCONNECT, PacketTypes.AUTH]),
- 33: (self.types.index("Two Byte Integer"),
- [PacketTypes.CONNECT, PacketTypes.CONNACK]),
- 34: (self.types.index("Two Byte Integer"),
- [PacketTypes.CONNECT, PacketTypes.CONNACK]),
- 35: (self.types.index("Two Byte Integer"), [PacketTypes.PUBLISH]),
- 36: (self.types.index("Byte"), [PacketTypes.CONNACK]),
- 37: (self.types.index("Byte"), [PacketTypes.CONNACK]),
- 38: (self.types.index("UTF-8 String Pair"),
- [PacketTypes.CONNECT, PacketTypes.CONNACK,
- PacketTypes.PUBLISH, PacketTypes.PUBACK,
- PacketTypes.PUBREC, PacketTypes.PUBREL, PacketTypes.PUBCOMP,
- PacketTypes.SUBSCRIBE, PacketTypes.SUBACK,
- PacketTypes.UNSUBSCRIBE, PacketTypes.UNSUBACK,
- PacketTypes.DISCONNECT, PacketTypes.AUTH, PacketTypes.WILLMESSAGE]),
- 39: (self.types.index("Four Byte Integer"),
- [PacketTypes.CONNECT, PacketTypes.CONNACK]),
- 40: (self.types.index("Byte"), [PacketTypes.CONNACK]),
- 41: (self.types.index("Byte"), [PacketTypes.CONNACK]),
- 42: (self.types.index("Byte"), [PacketTypes.CONNACK]),
- }
-
- def allowsMultiple(self, compressedName):
- return self.getIdentFromName(compressedName) in [11, 38]
-
- def getIdentFromName(self, compressedName):
- # return the identifier corresponding to the property name
- result = -1
- for name in self.names.keys():
- if compressedName == name.replace(' ', ''):
- result = self.names[name]
- break
- return result
-
- def __setattr__(self, name, value):
- name = name.replace(' ', '')
- privateVars = ["packetType", "types", "names", "properties"]
- if name in privateVars:
- object.__setattr__(self, name, value)
- else:
- # the name could have spaces in, or not. Remove spaces before assignment
- if name not in [aname.replace(' ', '') for aname in self.names.keys()]:
- raise MQTTException(
- f"Property name must be one of {self.names.keys()}")
- # check that this attribute applies to the packet type
- if self.packetType not in self.properties[self.getIdentFromName(name)][1]:
- raise MQTTException(f"Property {name} does not apply to packet type {PacketTypes.Names[self.packetType]}")
-
- # Check for forbidden values
- if not isinstance(value, list):
- if name in ["ReceiveMaximum", "TopicAlias"] \
- and (value < 1 or value > 65535):
-
- raise MQTTException(f"{name} property value must be in the range 1-65535")
- elif name in ["TopicAliasMaximum"] \
- and (value < 0 or value > 65535):
-
- raise MQTTException(f"{name} property value must be in the range 0-65535")
- elif name in ["MaximumPacketSize", "SubscriptionIdentifier"] \
- and (value < 1 or value > 268435455):
-
- raise MQTTException(f"{name} property value must be in the range 1-268435455")
- elif name in ["RequestResponseInformation", "RequestProblemInformation", "PayloadFormatIndicator"] \
- and (value != 0 and value != 1):
-
- raise MQTTException(
- f"{name} property value must be 0 or 1")
-
- if self.allowsMultiple(name):
- if not isinstance(value, list):
- value = [value]
- if hasattr(self, name):
- value = object.__getattribute__(self, name) + value
- object.__setattr__(self, name, value)
-
- def __str__(self):
- buffer = "["
- first = True
- for name in self.names.keys():
- compressedName = name.replace(' ', '')
- if hasattr(self, compressedName):
- if not first:
- buffer += ", "
- buffer += f"{compressedName} : {getattr(self, compressedName)}"
- first = False
- buffer += "]"
- return buffer
-
- def json(self):
- data = {}
- for name in self.names.keys():
- compressedName = name.replace(' ', '')
- if hasattr(self, compressedName):
- val = getattr(self, compressedName)
- if compressedName == 'CorrelationData' and isinstance(val, bytes):
- data[compressedName] = val.hex()
- else:
- data[compressedName] = val
- return data
-
- def isEmpty(self):
- rc = True
- for name in self.names.keys():
- compressedName = name.replace(' ', '')
- if hasattr(self, compressedName):
- rc = False
- break
- return rc
-
- def clear(self):
- for name in self.names.keys():
- compressedName = name.replace(' ', '')
- if hasattr(self, compressedName):
- delattr(self, compressedName)
-
- def writeProperty(self, identifier, type, value):
- buffer = b""
- buffer += VariableByteIntegers.encode(identifier) # identifier
- if type == self.types.index("Byte"): # value
- buffer += bytes([value])
- elif type == self.types.index("Two Byte Integer"):
- buffer += writeInt16(value)
- elif type == self.types.index("Four Byte Integer"):
- buffer += writeInt32(value)
- elif type == self.types.index("Variable Byte Integer"):
- buffer += VariableByteIntegers.encode(value)
- elif type == self.types.index("Binary Data"):
- buffer += writeBytes(value)
- elif type == self.types.index("UTF-8 Encoded String"):
- buffer += writeUTF(value)
- elif type == self.types.index("UTF-8 String Pair"):
- buffer += writeUTF(value[0]) + writeUTF(value[1])
- return buffer
-
- def pack(self):
- # serialize properties into buffer for sending over network
- buffer = b""
- for name in self.names.keys():
- compressedName = name.replace(' ', '')
- if hasattr(self, compressedName):
- identifier = self.getIdentFromName(compressedName)
- attr_type = self.properties[identifier][0]
- if self.allowsMultiple(compressedName):
- for prop in getattr(self, compressedName):
- buffer += self.writeProperty(identifier,
- attr_type, prop)
- else:
- buffer += self.writeProperty(identifier, attr_type,
- getattr(self, compressedName))
- return VariableByteIntegers.encode(len(buffer)) + buffer
-
- def readProperty(self, buffer, type, propslen):
- if type == self.types.index("Byte"):
- value = buffer[0]
- valuelen = 1
- elif type == self.types.index("Two Byte Integer"):
- value = readInt16(buffer)
- valuelen = 2
- elif type == self.types.index("Four Byte Integer"):
- value = readInt32(buffer)
- valuelen = 4
- elif type == self.types.index("Variable Byte Integer"):
- value, valuelen = VariableByteIntegers.decode(buffer)
- elif type == self.types.index("Binary Data"):
- value, valuelen = readBytes(buffer)
- elif type == self.types.index("UTF-8 Encoded String"):
- value, valuelen = readUTF(buffer, propslen)
- elif type == self.types.index("UTF-8 String Pair"):
- value, valuelen = readUTF(buffer, propslen)
- buffer = buffer[valuelen:] # strip the bytes used by the value
- value1, valuelen1 = readUTF(buffer, propslen - valuelen)
- value = (value, value1)
- valuelen += valuelen1
- return value, valuelen
-
- def getNameFromIdent(self, identifier):
- rc = None
- for name in self.names:
- if self.names[name] == identifier:
- rc = name
- return rc
-
- def unpack(self, buffer):
- self.clear()
- # deserialize properties into attributes from buffer received from network
- propslen, VBIlen = VariableByteIntegers.decode(buffer)
- buffer = buffer[VBIlen:] # strip the bytes used by the VBI
- propslenleft = propslen
- while propslenleft > 0: # properties length is 0 if there are none
- identifier, VBIlen2 = VariableByteIntegers.decode(
- buffer) # property identifier
- buffer = buffer[VBIlen2:] # strip the bytes used by the VBI
- propslenleft -= VBIlen2
- attr_type = self.properties[identifier][0]
- value, valuelen = self.readProperty(
- buffer, attr_type, propslenleft)
- buffer = buffer[valuelen:] # strip the bytes used by the value
- propslenleft -= valuelen
- propname = self.getNameFromIdent(identifier)
- compressedName = propname.replace(' ', '')
- if not self.allowsMultiple(compressedName) and hasattr(self, compressedName):
- raise MQTTException(
- f"Property '{property}' must not exist more than once")
- setattr(self, propname, value)
- return self, propslen + VBIlen
diff --git a/sbapp/mqtt/publish.py b/sbapp/mqtt/publish.py
deleted file mode 100644
index 333c190a..00000000
--- a/sbapp/mqtt/publish.py
+++ /dev/null
@@ -1,306 +0,0 @@
-# Copyright (c) 2014 Roger Light <roger@atchoo.org>
-#
-# All rights reserved. This program and the accompanying materials
-# are made available under the terms of the Eclipse Public License v2.0
-# and Eclipse Distribution License v1.0 which accompany this distribution.
-#
-# The Eclipse Public License is available at
-# http://www.eclipse.org/legal/epl-v20.html
-# and the Eclipse Distribution License is available at
-# http://www.eclipse.org/org/documents/edl-v10.php.
-#
-# Contributors:
-# Roger Light - initial API and implementation
-
-"""
-This module provides some helper functions to allow straightforward publishing
-of messages in a one-shot manner. In other words, they are useful for the
-situation where you have a single/multiple messages you want to publish to a
-broker, then disconnect and nothing else is required.
-"""
-from __future__ import annotations
-
-import collections
-from collections.abc import Iterable
-from typing import TYPE_CHECKING, Any, List, Tuple, Union
-
-from paho.mqtt.enums import CallbackAPIVersion, MQTTProtocolVersion
-from paho.mqtt.properties import Properties
-from paho.mqtt.reasoncodes import ReasonCode
-
-from .. import mqtt
-from . import client as paho
-
-if TYPE_CHECKING:
- try:
- from typing import NotRequired, Required, TypedDict # type: ignore
- except ImportError:
- from typing_extensions import NotRequired, Required, TypedDict
-
- try:
- from typing import Literal
- except ImportError:
- from typing_extensions import Literal # type: ignore
-
-
-
- class AuthParameter(TypedDict, total=False):
- username: Required[str]
- password: NotRequired[str]
-
-
- class TLSParameter(TypedDict, total=False):
- ca_certs: Required[str]
- certfile: NotRequired[str]
- keyfile: NotRequired[str]
- tls_version: NotRequired[int]
- ciphers: NotRequired[str]
- insecure: NotRequired[bool]
-
-
- class MessageDict(TypedDict, total=False):
- topic: Required[str]
- payload: NotRequired[paho.PayloadType]
- qos: NotRequired[int]
- retain: NotRequired[bool]
-
- MessageTuple = Tuple[str, paho.PayloadType, int, bool]
-
- MessagesList = List[Union[MessageDict, MessageTuple]]
-
-
-def _do_publish(client: paho.Client):
- """Internal function"""
-
- message = client._userdata.popleft()
-
- if isinstance(message, dict):
- client.publish(**message)
- elif isinstance(message, (tuple, list)):
- client.publish(*message)
- else:
- raise TypeError('message must be a dict, tuple, or list')
-
-
-def _on_connect(client: paho.Client, userdata: MessagesList, flags, reason_code, properties):
- """Internal v5 callback"""
- if reason_code == 0:
- if len(userdata) > 0:
- _do_publish(client)
- else:
- raise mqtt.MQTTException(paho.connack_string(reason_code))
-
-
-def _on_publish(
- client: paho.Client, userdata: collections.deque[MessagesList], mid: int, reason_codes: ReasonCode, properties: Properties,
-) -> None:
- """Internal callback"""
- #pylint: disable=unused-argument
-
- if len(userdata) == 0:
- client.disconnect()
- else:
- _do_publish(client)
-
-
-def multiple(
- msgs: MessagesList,
- hostname: str = "localhost",
- port: int = 1883,
- client_id: str = "",
- keepalive: int = 60,
- will: MessageDict | None = None,
- auth: AuthParameter | None = None,
- tls: TLSParameter | None = None,
- protocol: MQTTProtocolVersion = paho.MQTTv311,
- transport: Literal["tcp", "websockets"] = "tcp",
- proxy_args: Any | None = None,
-) -> None:
- """Publish multiple messages to a broker, then disconnect cleanly.
-
- This function creates an MQTT client, connects to a broker and publishes a
- list of messages. Once the messages have been delivered, it disconnects
- cleanly from the broker.
-
- :param msgs: a list of messages to publish. Each message is either a dict or a
- tuple.
-
- If a dict, only the topic must be present. Default values will be
- used for any missing arguments. The dict must be of the form:
-
- msg = {'topic':"<topic>", 'payload':"<payload>", 'qos':<qos>,
- 'retain':<retain>}
- topic must be present and may not be empty.
- If payload is "", None or not present then a zero length payload
- will be published.
- If qos is not present, the default of 0 is used.
- If retain is not present, the default of False is used.
-
- If a tuple, then it must be of the form:
- ("<topic>", "<payload>", qos, retain)
-
- :param str hostname: the address of the broker to connect to.
- Defaults to localhost.
-
- :param int port: the port to connect to the broker on. Defaults to 1883.
-
- :param str client_id: the MQTT client id to use. If "" or None, the Paho library will
- generate a client id automatically.
-
- :param int keepalive: the keepalive timeout value for the client. Defaults to 60
- seconds.
-
- :param will: a dict containing will parameters for the client: will = {'topic':
- "<topic>", 'payload':"<payload">, 'qos':<qos>, 'retain':<retain>}.
- Topic is required, all other parameters are optional and will
- default to None, 0 and False respectively.
- Defaults to None, which indicates no will should be used.
-
- :param auth: a dict containing authentication parameters for the client:
- auth = {'username':"<username>", 'password':"<password>"}
- Username is required, password is optional and will default to None
- if not provided.
- Defaults to None, which indicates no authentication is to be used.
-
- :param tls: a dict containing TLS configuration parameters for the client:
- dict = {'ca_certs':"<ca_certs>", 'certfile':"<certfile>",
- 'keyfile':"<keyfile>", 'tls_version':"<tls_version>",
- 'ciphers':"<ciphers">, 'insecure':"<bool>"}
- ca_certs is required, all other parameters are optional and will
- default to None if not provided, which results in the client using
- the default behaviour - see the paho.mqtt.client documentation.
- Alternatively, tls input can be an SSLContext object, which will be
- processed using the tls_set_context method.
- Defaults to None, which indicates that TLS should not be used.
-
- :param str transport: set to "tcp" to use the default setting of transport which is
- raw TCP. Set to "websockets" to use WebSockets as the transport.
-
- :param proxy_args: a dictionary that will be given to the client.
- """
-
- if not isinstance(msgs, Iterable):
- raise TypeError('msgs must be an iterable')
- if len(msgs) == 0:
- raise ValueError('msgs is empty')
-
- client = paho.Client(
- CallbackAPIVersion.VERSION2,
- client_id=client_id,
- userdata=collections.deque(msgs),
- protocol=protocol,
- transport=transport,
- )
-
- client.enable_logger()
- client.on_publish = _on_publish
- client.on_connect = _on_connect # type: ignore
-
- if proxy_args is not None:
- client.proxy_set(**proxy_args)
-
- if auth:
- username = auth.get('username')
- if username:
- password = auth.get('password')
- client.username_pw_set(username, password)
- else:
- raise KeyError("The 'username' key was not found, this is "
- "required for auth")
-
- if will is not None:
- client.will_set(**will)
-
- if tls is not None:
- if isinstance(tls, dict):
- insecure = tls.pop('insecure', False)
- # mypy don't get that tls no longer contains the key insecure
- client.tls_set(**tls) # type: ignore[misc]
- if insecure:
- # Must be set *after* the `client.tls_set()` call since it sets
- # up the SSL context that `client.tls_insecure_set` alters.
- client.tls_insecure_set(insecure)
- else:
- # Assume input is SSLContext object
- client.tls_set_context(tls)
-
- client.connect(hostname, port, keepalive)
- client.loop_forever()
-
-
-def single(
- topic: str,
- payload: paho.PayloadType = None,
- qos: int = 0,
- retain: bool = False,
- hostname: str = "localhost",
- port: int = 1883,
- client_id: str = "",
- keepalive: int = 60,
- will: MessageDict | None = None,
- auth: AuthParameter | None = None,
- tls: TLSParameter | None = None,
- protocol: MQTTProtocolVersion = paho.MQTTv311,
- transport: Literal["tcp", "websockets"] = "tcp",
- proxy_args: Any | None = None,
-) -> None:
- """Publish a single message to a broker, then disconnect cleanly.
-
- This function creates an MQTT client, connects to a broker and publishes a
- single message. Once the message has been delivered, it disconnects cleanly
- from the broker.
-
- :param str topic: the only required argument must be the topic string to which the
- payload will be published.
-
- :param payload: the payload to be published. If "" or None, a zero length payload
- will be published.
-
- :param int qos: the qos to use when publishing, default to 0.
-
- :param bool retain: set the message to be retained (True) or not (False).
-
- :param str hostname: the address of the broker to connect to.
- Defaults to localhost.
-
- :param int port: the port to connect to the broker on. Defaults to 1883.
-
- :param str client_id: the MQTT client id to use. If "" or None, the Paho library will
- generate a client id automatically.
-
- :param int keepalive: the keepalive timeout value for the client. Defaults to 60
- seconds.
-
- :param will: a dict containing will parameters for the client: will = {'topic':
- "<topic>", 'payload':"<payload">, 'qos':<qos>, 'retain':<retain>}.
- Topic is required, all other parameters are optional and will
- default to None, 0 and False respectively.
- Defaults to None, which indicates no will should be used.
-
- :param auth: a dict containing authentication parameters for the client:
- Username is required, password is optional and will default to None
- auth = {'username':"<username>", 'password':"<password>"}
- if not provided.
- Defaults to None, which indicates no authentication is to be used.
-
- :param tls: a dict containing TLS configuration parameters for the client:
- dict = {'ca_certs':"<ca_certs>", 'certfile':"<certfile>",
- 'keyfile':"<keyfile>", 'tls_version':"<tls_version>",
- 'ciphers':"<ciphers">, 'insecure':"<bool>"}
- ca_certs is required, all other parameters are optional and will
- default to None if not provided, which results in the client using
- the default behaviour - see the paho.mqtt.client documentation.
- Defaults to None, which indicates that TLS should not be used.
- Alternatively, tls input can be an SSLContext object, which will be
- processed using the tls_set_context method.
-
- :param transport: set to "tcp" to use the default setting of transport which is
- raw TCP. Set to "websockets" to use WebSockets as the transport.
-
- :param proxy_args: a dictionary that will be given to the client.
- """
-
- msg: MessageDict = {'topic':topic, 'payload':payload, 'qos':qos, 'retain':retain}
-
- multiple([msg], hostname, port, client_id, keepalive, will, auth, tls,
- protocol, transport, proxy_args)
diff --git a/sbapp/mqtt/py.typed b/sbapp/mqtt/py.typed
deleted file mode 100644
index e69de29b..00000000
diff --git a/sbapp/mqtt/reasoncodes.py b/sbapp/mqtt/reasoncodes.py
deleted file mode 100644
index 243ac96f..00000000
--- a/sbapp/mqtt/reasoncodes.py
+++ /dev/null
@@ -1,223 +0,0 @@
-# *******************************************************************
-# Copyright (c) 2017, 2019 IBM Corp.
-#
-# All rights reserved. This program and the accompanying materials
-# are made available under the terms of the Eclipse Public License v2.0
-# and Eclipse Distribution License v1.0 which accompany this distribution.
-#
-# The Eclipse Public License is available at
-# http://www.eclipse.org/legal/epl-v20.html
-# and the Eclipse Distribution License is available at
-# http://www.eclipse.org/org/documents/edl-v10.php.
-#
-# Contributors:
-# Ian Craggs - initial implementation and/or documentation
-# *******************************************************************
-
-import functools
-import warnings
-from typing import Any
-
-from .packettypes import PacketTypes
-
-
-@functools.total_ordering
-class ReasonCode:
- """MQTT version 5.0 reason codes class.
-
- See ReasonCode.names for a list of possible numeric values along with their
- names and the packets to which they apply.
-
- """
-
- def __init__(self, packetType: int, aName: str ="Success", identifier: int =-1):
- """
- packetType: the type of the packet, such as PacketTypes.CONNECT that
- this reason code will be used with. Some reason codes have different
- names for the same identifier when used a different packet type.
-
- aName: the String name of the reason code to be created. Ignored
- if the identifier is set.
-
- identifier: an integer value of the reason code to be created.
-
- """
-
- self.packetType = packetType
- self.names = {
- 0: {"Success": [PacketTypes.CONNACK, PacketTypes.PUBACK,
- PacketTypes.PUBREC, PacketTypes.PUBREL, PacketTypes.PUBCOMP,
- PacketTypes.UNSUBACK, PacketTypes.AUTH],
- "Normal disconnection": [PacketTypes.DISCONNECT],
- "Granted QoS 0": [PacketTypes.SUBACK]},
- 1: {"Granted QoS 1": [PacketTypes.SUBACK]},
- 2: {"Granted QoS 2": [PacketTypes.SUBACK]},
- 4: {"Disconnect with will message": [PacketTypes.DISCONNECT]},
- 16: {"No matching subscribers":
- [PacketTypes.PUBACK, PacketTypes.PUBREC]},
- 17: {"No subscription found": [PacketTypes.UNSUBACK]},
- 24: {"Continue authentication": [PacketTypes.AUTH]},
- 25: {"Re-authenticate": [PacketTypes.AUTH]},
- 128: {"Unspecified error": [PacketTypes.CONNACK, PacketTypes.PUBACK,
- PacketTypes.PUBREC, PacketTypes.SUBACK, PacketTypes.UNSUBACK,
- PacketTypes.DISCONNECT], },
- 129: {"Malformed packet":
- [PacketTypes.CONNACK, PacketTypes.DISCONNECT]},
- 130: {"Protocol error":
- [PacketTypes.CONNACK, PacketTypes.DISCONNECT]},
- 131: {"Implementation specific error": [PacketTypes.CONNACK,
- PacketTypes.PUBACK, PacketTypes.PUBREC, PacketTypes.SUBACK,
- PacketTypes.UNSUBACK, PacketTypes.DISCONNECT], },
- 132: {"Unsupported protocol version": [PacketTypes.CONNACK]},
- 133: {"Client identifier not valid": [PacketTypes.CONNACK]},
- 134: {"Bad user name or password": [PacketTypes.CONNACK]},
- 135: {"Not authorized": [PacketTypes.CONNACK, PacketTypes.PUBACK,
- PacketTypes.PUBREC, PacketTypes.SUBACK, PacketTypes.UNSUBACK,
- PacketTypes.DISCONNECT], },
- 136: {"Server unavailable": [PacketTypes.CONNACK]},
- 137: {"Server busy": [PacketTypes.CONNACK, PacketTypes.DISCONNECT]},
- 138: {"Banned": [PacketTypes.CONNACK]},
- 139: {"Server shutting down": [PacketTypes.DISCONNECT]},
- 140: {"Bad authentication method":
- [PacketTypes.CONNACK, PacketTypes.DISCONNECT]},
- 141: {"Keep alive timeout": [PacketTypes.DISCONNECT]},
- 142: {"Session taken over": [PacketTypes.DISCONNECT]},
- 143: {"Topic filter invalid":
- [PacketTypes.SUBACK, PacketTypes.UNSUBACK, PacketTypes.DISCONNECT]},
- 144: {"Topic name invalid":
- [PacketTypes.CONNACK, PacketTypes.PUBACK,
- PacketTypes.PUBREC, PacketTypes.DISCONNECT]},
- 145: {"Packet identifier in use":
- [PacketTypes.PUBACK, PacketTypes.PUBREC,
- PacketTypes.SUBACK, PacketTypes.UNSUBACK]},
- 146: {"Packet identifier not found":
- [PacketTypes.PUBREL, PacketTypes.PUBCOMP]},
- 147: {"Receive maximum exceeded": [PacketTypes.DISCONNECT]},
- 148: {"Topic alias invalid": [PacketTypes.DISCONNECT]},
- 149: {"Packet too large": [PacketTypes.CONNACK, PacketTypes.DISCONNECT]},
- 150: {"Message rate too high": [PacketTypes.DISCONNECT]},
- 151: {"Quota exceeded": [PacketTypes.CONNACK, PacketTypes.PUBACK,
- PacketTypes.PUBREC, PacketTypes.SUBACK, PacketTypes.DISCONNECT], },
- 152: {"Administrative action": [PacketTypes.DISCONNECT]},
- 153: {"Payload format invalid":
- [PacketTypes.PUBACK, PacketTypes.PUBREC, PacketTypes.DISCONNECT]},
- 154: {"Retain not supported":
- [PacketTypes.CONNACK, PacketTypes.DISCONNECT]},
- 155: {"QoS not supported":
- [PacketTypes.CONNACK, PacketTypes.DISCONNECT]},
- 156: {"Use another server":
- [PacketTypes.CONNACK, PacketTypes.DISCONNECT]},
- 157: {"Server moved":
- [PacketTypes.CONNACK, PacketTypes.DISCONNECT]},
- 158: {"Shared subscription not supported":
- [PacketTypes.SUBACK, PacketTypes.DISCONNECT]},
- 159: {"Connection rate exceeded":
- [PacketTypes.CONNACK, PacketTypes.DISCONNECT]},
- 160: {"Maximum connect time":
- [PacketTypes.DISCONNECT]},
- 161: {"Subscription identifiers not supported":
- [PacketTypes.SUBACK, PacketTypes.DISCONNECT]},
- 162: {"Wildcard subscription not supported":
- [PacketTypes.SUBACK, PacketTypes.DISCONNECT]},
- }
- if identifier == -1:
- if packetType == PacketTypes.DISCONNECT and aName == "Success":
- aName = "Normal disconnection"
- self.set(aName)
- else:
- self.value = identifier
- self.getName() # check it's good
-
- def __getName__(self, packetType, identifier):
- """
- Get the reason code string name for a specific identifier.
- The name can vary by packet type for the same identifier, which
- is why the packet type is also required.
-
- Used when displaying the reason code.
- """
- if identifier not in self.names:
- raise KeyError(identifier)
- names = self.names[identifier]
- namelist = [name for name in names.keys() if packetType in names[name]]
- if len(namelist) != 1:
- raise ValueError(f"Expected exactly one name, found {namelist!r}")
- return namelist[0]
-
- def getId(self, name):
- """
- Get the numeric id corresponding to a reason code name.
-
- Used when setting the reason code for a packetType
- check that only valid codes for the packet are set.
- """
- for code in self.names.keys():
- if name in self.names[code].keys():
- if self.packetType in self.names[code][name]:
- return code
- raise KeyError(f"Reason code name not found: {name}")
-
- def set(self, name):
- self.value = self.getId(name)
-
- def unpack(self, buffer):
- c = buffer[0]
- name = self.__getName__(self.packetType, c)
- self.value = self.getId(name)
- return 1
-
- def getName(self):
- """Returns the reason code name corresponding to the numeric value which is set.
- """
- return self.__getName__(self.packetType, self.value)
-
- def __eq__(self, other):
- if isinstance(other, int):
- return self.value == other
- if isinstance(other, str):
- return other == str(self)
- if isinstance(other, ReasonCode):
- return self.value == other.value
- return False
-
- def __lt__(self, other):
- if isinstance(other, int):
- return self.value < other
- if isinstance(other, ReasonCode):
- return self.value < other.value
- return NotImplemented
-
- def __repr__(self):
- try:
- packet_name = PacketTypes.Names[self.packetType]
- except IndexError:
- packet_name = "Unknown"
-
- return f"ReasonCode({packet_name}, {self.getName()!r})"
-
- def __str__(self):
- return self.getName()
-
- def json(self):
- return self.getName()
-
- def pack(self):
- return bytearray([self.value])
-
- @property
- def is_failure(self) -> bool:
- return self.value >= 0x80
-
-
-class _CompatibilityIsInstance(type):
- def __instancecheck__(self, other: Any) -> bool:
- return isinstance(other, ReasonCode)
-
-
-class ReasonCodes(ReasonCode, metaclass=_CompatibilityIsInstance):
- def __init__(self, *args, **kwargs):
- warnings.warn("ReasonCodes is deprecated, use ReasonCode (singular) instead",
- category=DeprecationWarning,
- stacklevel=2,
- )
- super().__init__(*args, **kwargs)
diff --git a/sbapp/mqtt/subscribe.py b/sbapp/mqtt/subscribe.py
deleted file mode 100644
index b6c80f44..00000000
--- a/sbapp/mqtt/subscribe.py
+++ /dev/null
@@ -1,281 +0,0 @@
-# Copyright (c) 2016 Roger Light <roger@atchoo.org>
-#
-# All rights reserved. This program and the accompanying materials
-# are made available under the terms of the Eclipse Public License v2.0
-# and Eclipse Distribution License v1.0 which accompany this distribution.
-#
-# The Eclipse Public License is available at
-# http://www.eclipse.org/legal/epl-v20.html
-# and the Eclipse Distribution License is available at
-# http://www.eclipse.org/org/documents/edl-v10.php.
-#
-# Contributors:
-# Roger Light - initial API and implementation
-
-"""
-This module provides some helper functions to allow straightforward subscribing
-to topics and retrieving messages. The two functions are simple(), which
-returns one or messages matching a set of topics, and callback() which allows
-you to pass a callback for processing of messages.
-"""
-
-from .. import mqtt
-from . import client as paho
-
-
-def _on_connect(client, userdata, flags, reason_code, properties):
- """Internal callback"""
- if reason_code != 0:
- raise mqtt.MQTTException(paho.connack_string(reason_code))
-
- if isinstance(userdata['topics'], list):
- for topic in userdata['topics']:
- client.subscribe(topic, userdata['qos'])
- else:
- client.subscribe(userdata['topics'], userdata['qos'])
-
-
-def _on_message_callback(client, userdata, message):
- """Internal callback"""
- userdata['callback'](client, userdata['userdata'], message)
-
-
-def _on_message_simple(client, userdata, message):
- """Internal callback"""
-
- if userdata['msg_count'] == 0:
- return
-
- # Don't process stale retained messages if 'retained' was false
- if message.retain and not userdata['retained']:
- return
-
- userdata['msg_count'] = userdata['msg_count'] - 1
-
- if userdata['messages'] is None and userdata['msg_count'] == 0:
- userdata['messages'] = message
- client.disconnect()
- return
-
- userdata['messages'].append(message)
- if userdata['msg_count'] == 0:
- client.disconnect()
-
-
-def callback(callback, topics, qos=0, userdata=None, hostname="localhost",
- port=1883, client_id="", keepalive=60, will=None, auth=None,
- tls=None, protocol=paho.MQTTv311, transport="tcp",
- clean_session=True, proxy_args=None):
- """Subscribe to a list of topics and process them in a callback function.
-
- This function creates an MQTT client, connects to a broker and subscribes
- to a list of topics. Incoming messages are processed by the user provided
- callback. This is a blocking function and will never return.
-
- :param callback: function with the same signature as `on_message` for
- processing the messages received.
-
- :param topics: either a string containing a single topic to subscribe to, or a
- list of topics to subscribe to.
-
- :param int qos: the qos to use when subscribing. This is applied to all topics.
-
- :param userdata: passed to the callback
-
- :param str hostname: the address of the broker to connect to.
- Defaults to localhost.
-
- :param int port: the port to connect to the broker on. Defaults to 1883.
-
- :param str client_id: the MQTT client id to use. If "" or None, the Paho library will
- generate a client id automatically.
-
- :param int keepalive: the keepalive timeout value for the client. Defaults to 60
- seconds.
-
- :param will: a dict containing will parameters for the client: will = {'topic':
- "<topic>", 'payload':"<payload">, 'qos':<qos>, 'retain':<retain>}.
- Topic is required, all other parameters are optional and will
- default to None, 0 and False respectively.
-
- Defaults to None, which indicates no will should be used.
-
- :param auth: a dict containing authentication parameters for the client:
- auth = {'username':"<username>", 'password':"<password>"}
- Username is required, password is optional and will default to None
- if not provided.
- Defaults to None, which indicates no authentication is to be used.
-
- :param tls: a dict containing TLS configuration parameters for the client:
- dict = {'ca_certs':"<ca_certs>", 'certfile':"<certfile>",
- 'keyfile':"<keyfile>", 'tls_version':"<tls_version>",
- 'ciphers':"<ciphers">, 'insecure':"<bool>"}
- ca_certs is required, all other parameters are optional and will
- default to None if not provided, which results in the client using
- the default behaviour - see the paho.mqtt.client documentation.
- Alternatively, tls input can be an SSLContext object, which will be
- processed using the tls_set_context method.
- Defaults to None, which indicates that TLS should not be used.
-
- :param str transport: set to "tcp" to use the default setting of transport which is
- raw TCP. Set to "websockets" to use WebSockets as the transport.
-
- :param clean_session: a boolean that determines the client type. If True,
- the broker will remove all information about this client
- when it disconnects. If False, the client is a persistent
- client and subscription information and queued messages
- will be retained when the client disconnects.
- Defaults to True.
-
- :param proxy_args: a dictionary that will be given to the client.
- """
-
- if qos < 0 or qos > 2:
- raise ValueError('qos must be in the range 0-2')
-
- callback_userdata = {
- 'callback':callback,
- 'topics':topics,
- 'qos':qos,
- 'userdata':userdata}
-
- client = paho.Client(
- paho.CallbackAPIVersion.VERSION2,
- client_id=client_id,
- userdata=callback_userdata,
- protocol=protocol,
- transport=transport,
- clean_session=clean_session,
- )
- client.enable_logger()
-
- client.on_message = _on_message_callback
- client.on_connect = _on_connect
-
- if proxy_args is not None:
- client.proxy_set(**proxy_args)
-
- if auth:
- username = auth.get('username')
- if username:
- password = auth.get('password')
- client.username_pw_set(username, password)
- else:
- raise KeyError("The 'username' key was not found, this is "
- "required for auth")
-
- if will is not None:
- client.will_set(**will)
-
- if tls is not None:
- if isinstance(tls, dict):
- insecure = tls.pop('insecure', False)
- client.tls_set(**tls)
- if insecure:
- # Must be set *after* the `client.tls_set()` call since it sets
- # up the SSL context that `client.tls_insecure_set` alters.
- client.tls_insecure_set(insecure)
- else:
- # Assume input is SSLContext object
- client.tls_set_context(tls)
-
- client.connect(hostname, port, keepalive)
- client.loop_forever()
-
-
-def simple(topics, qos=0, msg_count=1, retained=True, hostname="localhost",
- port=1883, client_id="", keepalive=60, will=None, auth=None,
- tls=None, protocol=paho.MQTTv311, transport="tcp",
- clean_session=True, proxy_args=None):
- """Subscribe to a list of topics and return msg_count messages.
-
- This function creates an MQTT client, connects to a broker and subscribes
- to a list of topics. Once "msg_count" messages have been received, it
- disconnects cleanly from the broker and returns the messages.
-
- :param topics: either a string containing a single topic to subscribe to, or a
- list of topics to subscribe to.
-
- :param int qos: the qos to use when subscribing. This is applied to all topics.
-
- :param int msg_count: the number of messages to retrieve from the broker.
- if msg_count == 1 then a single MQTTMessage will be returned.
- if msg_count > 1 then a list of MQTTMessages will be returned.
-
- :param bool retained: If set to True, retained messages will be processed the same as
- non-retained messages. If set to False, retained messages will
- be ignored. This means that with retained=False and msg_count=1,
- the function will return the first message received that does
- not have the retained flag set.
-
- :param str hostname: the address of the broker to connect to.
- Defaults to localhost.
-
- :param int port: the port to connect to the broker on. Defaults to 1883.
-
- :param str client_id: the MQTT client id to use. If "" or None, the Paho library will
- generate a client id automatically.
-
- :param int keepalive: the keepalive timeout value for the client. Defaults to 60
- seconds.
-
- :param will: a dict containing will parameters for the client: will = {'topic':
- "<topic>", 'payload':"<payload">, 'qos':<qos>, 'retain':<retain>}.
- Topic is required, all other parameters are optional and will
- default to None, 0 and False respectively.
- Defaults to None, which indicates no will should be used.
-
- :param auth: a dict containing authentication parameters for the client:
- auth = {'username':"<username>", 'password':"<password>"}
- Username is required, password is optional and will default to None
- if not provided.
- Defaults to None, which indicates no authentication is to be used.
-
- :param tls: a dict containing TLS configuration parameters for the client:
- dict = {'ca_certs':"<ca_certs>", 'certfile':"<certfile>",
- 'keyfile':"<keyfile>", 'tls_version':"<tls_version>",
- 'ciphers':"<ciphers">, 'insecure':"<bool>"}
- ca_certs is required, all other parameters are optional and will
- default to None if not provided, which results in the client using
- the default behaviour - see the paho.mqtt.client documentation.
- Alternatively, tls input can be an SSLContext object, which will be
- processed using the tls_set_context method.
- Defaults to None, which indicates that TLS should not be used.
-
- :param protocol: the MQTT protocol version to use. Defaults to MQTTv311.
-
- :param transport: set to "tcp" to use the default setting of transport which is
- raw TCP. Set to "websockets" to use WebSockets as the transport.
-
- :param clean_session: a boolean that determines the client type. If True,
- the broker will remove all information about this client
- when it disconnects. If False, the client is a persistent
- client and subscription information and queued messages
- will be retained when the client disconnects.
- Defaults to True. If protocol is MQTTv50, clean_session
- is ignored.
-
- :param proxy_args: a dictionary that will be given to the client.
- """
-
- if msg_count < 1:
- raise ValueError('msg_count must be > 0')
-
- # Set ourselves up to return a single message if msg_count == 1, or a list
- # if > 1.
- if msg_count == 1:
- messages = None
- else:
- messages = []
-
- # Ignore clean_session if protocol is MQTTv50, otherwise Client will raise
- if protocol == paho.MQTTv5:
- clean_session = None
-
- userdata = {'retained':retained, 'msg_count':msg_count, 'messages':messages}
-
- callback(_on_message_simple, topics, qos, userdata, hostname, port,
- client_id, keepalive, will, auth, tls, protocol, transport,
- clean_session, proxy_args)
-
- return userdata['messages']
diff --git a/sbapp/mqtt/subscribeoptions.py b/sbapp/mqtt/subscribeoptions.py
deleted file mode 100644
index 7e0605de..00000000
--- a/sbapp/mqtt/subscribeoptions.py
+++ /dev/null
@@ -1,113 +0,0 @@
-"""
-*******************************************************************
- Copyright (c) 2017, 2019 IBM Corp.
-
- All rights reserved. This program and the accompanying materials
- are made available under the terms of the Eclipse Public License v2.0
- and Eclipse Distribution License v1.0 which accompany this distribution.
-
- The Eclipse Public License is available at
- http://www.eclipse.org/legal/epl-v20.html
- and the Eclipse Distribution License is available at
- http://www.eclipse.org/org/documents/edl-v10.php.
-
- Contributors:
- Ian Craggs - initial implementation and/or documentation
-*******************************************************************
-"""
-
-
-
-class MQTTException(Exception):
- pass
-
-
-class SubscribeOptions:
- """The MQTT v5.0 subscribe options class.
-
- The options are:
- qos: As in MQTT v3.1.1.
- noLocal: True or False. If set to True, the subscriber will not receive its own publications.
- retainAsPublished: True or False. If set to True, the retain flag on received publications will be as set
- by the publisher.
- retainHandling: RETAIN_SEND_ON_SUBSCRIBE, RETAIN_SEND_IF_NEW_SUB or RETAIN_DO_NOT_SEND
- Controls when the broker should send retained messages:
- - RETAIN_SEND_ON_SUBSCRIBE: on any successful subscribe request
- - RETAIN_SEND_IF_NEW_SUB: only if the subscribe request is new
- - RETAIN_DO_NOT_SEND: never send retained messages
- """
-
- # retain handling options
- RETAIN_SEND_ON_SUBSCRIBE, RETAIN_SEND_IF_NEW_SUB, RETAIN_DO_NOT_SEND = range(
- 0, 3)
-
- def __init__(
- self,
- qos: int = 0,
- noLocal: bool = False,
- retainAsPublished: bool = False,
- retainHandling: int = RETAIN_SEND_ON_SUBSCRIBE,
- ):
- """
- qos: 0, 1 or 2. 0 is the default.
- noLocal: True or False. False is the default and corresponds to MQTT v3.1.1 behavior.
- retainAsPublished: True or False. False is the default and corresponds to MQTT v3.1.1 behavior.
- retainHandling: RETAIN_SEND_ON_SUBSCRIBE, RETAIN_SEND_IF_NEW_SUB or RETAIN_DO_NOT_SEND
- RETAIN_SEND_ON_SUBSCRIBE is the default and corresponds to MQTT v3.1.1 behavior.
- """
- object.__setattr__(self, "names",
- ["QoS", "noLocal", "retainAsPublished", "retainHandling"])
- self.QoS = qos # bits 0,1
- self.noLocal = noLocal # bit 2
- self.retainAsPublished = retainAsPublished # bit 3
- self.retainHandling = retainHandling # bits 4 and 5: 0, 1 or 2
- if self.retainHandling not in (0, 1, 2):
- raise AssertionError(f"Retain handling should be 0, 1 or 2, not {self.retainHandling}")
- if self.QoS not in (0, 1, 2):
- raise AssertionError(f"QoS should be 0, 1 or 2, not {self.QoS}")
-
- def __setattr__(self, name, value):
- if name not in self.names:
- raise MQTTException(
- f"{name} Attribute name must be one of {self.names}")
- object.__setattr__(self, name, value)
-
- def pack(self):
- if self.retainHandling not in (0, 1, 2):
- raise AssertionError(f"Retain handling should be 0, 1 or 2, not {self.retainHandling}")
- if self.QoS not in (0, 1, 2):
- raise AssertionError(f"QoS should be 0, 1 or 2, not {self.QoS}")
- noLocal = 1 if self.noLocal else 0
- retainAsPublished = 1 if self.retainAsPublished else 0
- data = [(self.retainHandling << 4) | (retainAsPublished << 3) |
- (noLocal << 2) | self.QoS]
- return bytes(data)
-
- def unpack(self, buffer):
- b0 = buffer[0]
- self.retainHandling = ((b0 >> 4) & 0x03)
- self.retainAsPublished = True if ((b0 >> 3) & 0x01) == 1 else False
- self.noLocal = True if ((b0 >> 2) & 0x01) == 1 else False
- self.QoS = (b0 & 0x03)
- if self.retainHandling not in (0, 1, 2):
- raise AssertionError(f"Retain handling should be 0, 1 or 2, not {self.retainHandling}")
- if self.QoS not in (0, 1, 2):
- raise AssertionError(f"QoS should be 0, 1 or 2, not {self.QoS}")
- return 1
-
- def __repr__(self):
- return str(self)
-
- def __str__(self):
- return "{QoS="+str(self.QoS)+", noLocal="+str(self.noLocal) +\
- ", retainAsPublished="+str(self.retainAsPublished) +\
- ", retainHandling="+str(self.retainHandling)+"}"
-
- def json(self):
- data = {
- "QoS": self.QoS,
- "noLocal": self.noLocal,
- "retainAsPublished": self.retainAsPublished,
- "retainHandling": self.retainHandling,
- }
- return data
diff --git a/sbapp/sideband/core.py b/sbapp/sideband/core.py
index ce5547df..fa431644 100644
--- a/sbapp/sideband/core.py
+++ b/sbapp/sideband/core.py
@@ -3228,9 +3228,10 @@ class SidebandCore():
if self.config["telemetry_enabled"] == True:
self.update_telemeter_config()
if self.telemeter != None:
- def mqtt_job():
- self.mqtt_handle_telemetry(self.lxmf_destination.hash, self.telemeter.packed())
- threading.Thread(target=mqtt_job, daemon=True).start()
+ if self.config["telemetry_to_mqtt"]:
+ def mqtt_job():
+ self.mqtt_handle_telemetry(self.lxmf_destination.hash, self.telemeter.packed())
+ threading.Thread(target=mqtt_job, daemon=True).start()
return self.telemeter.read_all()
else:
return {}
diff --git a/sbapp/sideband/mqtt.py b/sbapp/sideband/mqtt.py
index 0d698167..0d4837ef 100644
--- a/sbapp/sideband/mqtt.py
+++ b/sbapp/sideband/mqtt.py
@@ -2,9 +2,13 @@ import RNS
import time
import threading
from collections import deque
-from sbapp.mqtt import client as mqtt
from .sense import Telemeter, Commands
+if RNS.vendor.platformutils.get_platform() == "android":
+ import pmqtt.client as mqtt
+else:
+ from sbapp.pmqtt import client as mqtt
+
class MQTT():
QUEUE_MAXLEN = 65536
SCHEDULER_SLEEP = 1
diff --git a/sbapp/sideband/sense.py b/sbapp/sideband/sense.py
index 77164170..1aa5a9bf 100644
--- a/sbapp/sideband/sense.py
+++ b/sbapp/sideband/sense.py
@@ -2542,7 +2542,6 @@ class RNSTransport(Sensor):
rss = ifstats.pop("rss")
if self.last_update == 0:
- RNS.log("NO CALC DIFF")
rxs = ifstats["rxs"]
txs = ifstats["txs"]
else:
@@ -2551,9 +2550,6 @@ class RNSTransport(Sensor):
txd = ifstats["txb"] - self._last_traffic_txb
rxs = (rxd/td)*8
txs = (txd/td)*8
- RNS.log(f"CALC DIFFS: td={td}, rxd={rxd}, txd={txd}")
- RNS.log(f" rxs={rxs}, txs={txs}")
-
self._last_traffic_rxb = ifstats["rxb"]
self._last_traffic_txb = ifstats["txb"]
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────